How to Integrate Databricks Unity Catalog for Enterprise Data Ingestion with Semantica

Semantica's Databricks ingestion module enables direct extraction of data, schema, and lineage from Unity Catalog workspaces using the DatabricksIngestor class, supporting both Personal Access Token and OAuth M2M service-principal authentication.

The semantica-agi/semantica repository ships a production-ready Databricks integration that bridges Unity Catalog's governance layer with Semantica's LLM pipelines. This integration allows enterprises to ingest petabyte-scale datasets while preserving metadata, schema information, and data lineage for knowledge-graph construction.

Architecture Overview

Semantica implements a three-layer architecture for Databricks Unity Catalog integration, defined in semantica/ingest/databricks_ingestor.py.

Connection Layer (DatabricksConnector): Handles low-level SQL connectivity via databricks-sql-connector and instantiates a WorkspaceClient from the databricks-sdk for Unity Catalog API operations. Located at line 79, this class manages authentication, connection pooling, and HTTP path resolution for SQL warehouses.

Ingestion Engine (DatabricksIngestor): Orchestrates table and query ingestion with pagination support, progress tracking, and result-set conversion. Implemented at line 103, this class provides methods for schema discovery, catalog enumeration, and lineage extraction through Unity Catalog's REST endpoints.

Data Model (DatabricksData): A lightweight dataclass (line 64) that encapsulates data, row_count, columns, and metadata, providing a standardized interface for converting query results into Semantica document objects.

Authentication and Connection Management

Semantica supports two authentication patterns for Databricks Unity Catalog: Personal Access Token (PAT) and OAuth M2M service-principal authentication.

Environment-Based Configuration

The DatabricksConnector class reads configuration from environment variables when explicit parameters are omitted:

export DATABRICKS_HOST="https://adb-<workspace>.azuredatabricks.net"
export DATABRICKS_HTTP_PATH="/sql/1.0/warehouses/<warehouse-id>"
export DATABRICKS_TOKEN="dapi-<personal-access-token>"  # For PAT auth

# OR for OAuth M2M:

export DATABRICKS_CLIENT_ID="my-sp-client-id"
export DATABRICKS_CLIENT_SECRET="my-sp-client-secret"

Authentication Implementation

For PAT authentication, the connector passes the access_token directly to databricks_sql.connect. For OAuth M2M, it constructs a credentials_provider callable using oauth_service_principal(Config(...)), which is required for databricks-sql-connector >= 2.5.

from semantica.ingest import DatabricksIngestor

# Personal Access Token authentication

ingestor = DatabricksIngestor(
    host="https://adb-xyz.azuredatabricks.net",
    token="dapi-abcdef123456",
    http_path="/sql/1.0/warehouses/1234",
)

# OAuth M2M service-principal authentication

ingestor = DatabricksIngestor(
    host="https://adb-xyz.azuredatabricks.net",
    client_id="sp-client-id",
    client_secret="sp-client-secret",
    http_path="/sql/1.0/warehouses/1234",
)

The connect() method creates a reusable SQL connection cached on the connector instance, while get_workspace_client() lazily initializes the Unity Catalog SDK client for metadata operations.

Ingesting Data from Unity Catalog

Table Ingestion

The ingest_table() method constructs fully-qualified, back-ticked table references (catalog.schema.table) via _escaped_table_ref and executes safe SELECT statements with optional filtering and ordering.

# Simple ingestion with row limits

data = ingestor.ingest_table("customers", limit=5000)

# Filtered ingestion with ordering

data = ingestor.ingest_table(
    "orders",
    where="order_date >= '2024-01-01'",
    order_by="order_date DESC",
)

Custom Query Ingestion

For complex transformations, ingest_query() accepts arbitrary SQL with parameterized values and batched fetching for memory-efficient processing of large result sets.

query_data = ingestor.ingest_query(
    "SELECT * FROM sales WHERE region = :region",
    parameters={"region": "EMEA"},
    batch_size=1000,
)

Metadata and Lineage Extraction

Schema Discovery

The integration exposes Unity Catalog's hierarchical namespace through list_catalogs(), list_schemas(), and list_tables(). Detailed schema information is retrieved via get_table_schema(), which calls client.tables.get(full_name=...) and normalizes column metadata.

# Enumerate available data assets

catalogs = ingestor.list_catalogs()
schemas = ingestor.list_schemas(catalog="main")
tables = ingestor.list_tables(catalog="main", schema="default")

# Retrieve detailed column information

schema = ingestor.get_table_schema("main.default.customers")
print([c["name"] for c in schema["columns"]])

Lineage Tracking

Semantica extracts both table-level and column-level lineage by invoking Unity Catalog's REST endpoints /api/2.0/lineage-tracking/table-lineage and /api/2.0/lineage-tracking/column-lineage.

lineage = ingestor.get_table_lineage(
    "main.default.orders",
    include_column_lineage=True
)

This enables integration with data governance workflows and knowledge-graph construction that tracks data provenance across your lakehouse.

Converting Data to Semantica Documents

The export_as_documents() method transforms DatabricksData instances into Semantica-compatible document dictionaries suitable for vector-store indexing or LLM prompting.

docs = ingestor.export_as_documents(
    data,
    id_field="customer_id",
    text_fields=["first_name", "last_name", "notes"],
    metadata_fields=["created_at", "region"]
)

# docs format: [{"id": "...", "text": "...", "metadata": {...}}, ...]

Complete Integration Example

The following example demonstrates end-to-end ingestion using context-manager semantics for automatic connection lifecycle management:

from semantica.ingest import DatabricksIngestor

with DatabricksIngestor(
    host="https://adb-abc.azuredatabricks.net",
    token="dapi-xyz123",
    http_path="/sql/1.0/warehouses/5678",
) as ingestor:

    # Discover available catalogs

    catalogs = ingestor.list_catalogs()
    print(f"Available catalogs: {catalogs}")
    
    # Ingest filtered table data

    customers = ingestor.ingest_table(
        "main.default.customers",
        where="status = 'ACTIVE'",
        limit=10_000,
    )
    print(f"Ingested {customers.row_count} rows")
    
    # Extract schema for validation

    schema = ingestor.get_table_schema("main.default.customers")
    
    # Retrieve lineage for governance

    lineage = ingestor.get_table_lineage("main.default.customers")
    
    # Convert to documents for vector storage

    docs = ingestor.export_as_documents(
        customers,
        id_field="customer_id",
        text_fields=["first_name", "last_name"],
    )
    # vector_store.add_documents(docs)

Context-manager usage ensures the SQL connection is opened once, reused across multiple operations, and cleanly closed on exit, as verified in the test suite's test_context_manager and test_context_manager_reuses_connection_across_calls cases.

Summary

• Three-layer architecture: DatabricksConnector manages connections, DatabricksIngestor orchestrates extraction, and DatabricksData standardizes outputs.
• Dual authentication: Supports both Personal Access Tokens and OAuth M2M service principals via databricks-sql-connector >= 2.5.
• Unity Catalog native: Direct integration with catalog listing, schema discovery, and lineage tracking REST endpoints.
• Production-ready: Context-manager support ensures efficient connection pooling, while batched query execution handles enterprise-scale datasets.
• LLM-ready output: Built-in export_as_documents() method bridges raw SQL results to Semantica vector stores.

Frequently Asked Questions

What authentication methods does Semantica support for Databricks Unity Catalog?

Semantica supports Personal Access Token (PAT) authentication for individual users and OAuth M2M service-principal authentication for production workloads. The DatabricksConnector class automatically detects the credential type based on whether token or client_id/client_secret parameters are provided, implementing the appropriate credentials_provider for the SQL connector.

How does Semantica handle large table ingestion from Databricks?

The ingest_query() method supports batched fetching via the batch_size parameter, which uses fetchmany() internally to stream large result sets without loading entire tables into memory. Additionally, the limit parameter on ingest_table() provides cap controls, and the connection pooling in DatabricksConnector reuses TCP connections across multiple chunked requests.

Can I extract column-level lineage information using Semantica's Databricks integration?

Yes. The get_table_lineage() method accepts an include_column_lineage=True parameter that queries the /api/2.0/lineage-tracking/column-lineage Unity Catalog REST endpoint. This returns upstream and downstream dependencies for individual columns, enabling fine-grained impact analysis and data governance workflows.

What is the difference between DatabricksConnector and DatabricksIngestor?

DatabricksConnector (semantica/ingest/databricks_ingestor.py#L79) is a low-level utility that manages SQL connections and the WorkspaceClient SDK instance. DatabricksIngestor (line 103) is a high-level orchestrator that uses the connector to execute queries, paginate results, track progress, and convert outputs to DatabricksData objects. Direct use of DatabricksConnector is rarely necessary unless implementing custom query patterns outside the ingestor's scope.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}