How to Connect Databricks Unity Catalog to Semantica for KG Ingestion

Use the DatabricksIngestor class to connect Databricks Unity Catalog to Semantica, authenticating via personal access token or OAuth M2M, then ingest tables or queries and export them as documents for knowledge-graph ingestion.

The Databricks Unity Catalog integration in Semantica provides a complete pipeline for extracting structured data from Apache Iceberg tables and loading it into a knowledge graph. According to the Semantica source code, the implementation centers on two core components: DatabricksConnector for low-level SQL connections and DatabricksIngestor for high-level data extraction and metadata operations.

---

Prerequisites and Authentication

Before ingesting data, you must authenticate against your Databricks workspace. The connector supports two methods:

• Personal Access Token (PAT) – simplest for individual users
• OAuth M2M – recommended for production service principals

The DatabricksConnector.connect() method handles credential resolution【/semantica/ingest/databricks_ingestor.py#L68-L82】. It checks for explicit parameters first, then falls back to standard Databricks environment variables (DATABRICKS_HOST, DATABRICKS_TOKEN, DATABRICKS_CLIENT_ID, DATABRICKS_CLIENT_SECRET).

---

Core Components

DatabricksConnector

The DatabricksConnector class manages:

• SQL warehouse connections via sql.connect()
• Unity Catalog client instantiation via WorkspaceClient
• Connection lifecycle (explicit disconnect() or automatic cleanup via context manager)

Authentication branching logic is implemented in connect():

# PAT path

connector = sql.connect(
    server_hostname=host,
    http_path=http_path,
    access_token=token,
)

# OAuth M2M path

connector = sql.connect(
    server_hostname=host,
    http_path=http_path,
    credentials_provider=credentials_provider,
)

DatabricksIngestor

The DatabricksIngestor wraps the connector and adds:

• Table and query ingestion with automatic JSON serialization
• Unity Catalog metadata exploration (catalogs, schemas, tables)
• Lineage retrieval via the Databricks Lineage API
• Progress tracking and structured error handling

The ingestor lazily initializes both the SQL connector and WorkspaceClient on first use, ensuring minimal resource overhead.

DatabricksData

Ingested results are returned as a DatabricksData dataclass【/semantica/ingest/databricks_ingestor.py#L64-L76】 containing:

• data: list of dictionary rows
• columns: list of column names
• row_count: total rows fetched
• query: source query or table reference
• catalog, schema, table: provenance metadata

---

Ingesting Tables from Unity Catalog

The primary workflow for Databricks Unity Catalog to Semantica KG ingestion is table extraction. The ingest_table() method【/semantica/ingest/databricks_ingestor.py#L15-L41】 handles pagination, progress tracking, and automatic type conversion.

from semantica.ingest import DatabricksIngestor

ingestor = DatabricksIngestor(
    host="https://adb-123.azuredatabricks.net",
    token="dapi-XXXXXXXXXXXXXXXX",
    http_path="/sql/1.0/warehouses/abc123",
    catalog="main",
    schema="default",
)

# Ingest with optional row limit

data = ingestor.ingest_table("customers", limit=10_000)

print(f"Fetched {data.row_count} rows with columns: {data.columns}")

For automatic connection cleanup, use the context manager pattern【/semantica/ingest/databricks_ingestor.py#L110-L117】:

with DatabricksIngestor(
    host="https://adb-123.azuredatabricks.net",
    token="dapi-XXXXXXXXXXXXXXXX",
    http_path="/sql/1.0/warehouses/abc123",
) as ingestor:
    data = ingestor.ingest_table("orders", limit=500)
    # connector.disconnect() called automatically

---

Executing Custom SQL Queries

For filtered or joined data, use ingest_query()【/semantica/ingest/databricks_ingestor.py#L52-L60】:

query = """
SELECT 
    o.order_id,
    c.customer_name,
    o.total_amount,
    o.order_date
FROM main.sales.orders o
JOIN main.sales.customers c ON o.customer_id = c.customer_id
WHERE o.order_date >= '2024-01-01'
ORDER BY o.total_amount DESC
LIMIT 1000
"""

data = ingestor.ingest_query(query)

All results are automatically converted to JSON-serializable Python native types, eliminating manual datetime/decimal handling.

---

Exploring Unity Catalog Metadata

The ingestor provides direct access to catalog metadata without writing SQL【/semantica/ingest/databricks_ingestor.py#L124-L140】:

# List all accessible catalogs

catalogs = ingestor.list_catalogs()

# List schemas within a catalog

schemas = ingestor.list_schemas(catalog="main")

# List tables in a specific schema

tables = ingestor.list_tables(catalog="main", schema="analytics")

---

Retrieving Schema and Lineage Information

Table Schema

Get column definitions with get_table_schema()【/semantica/ingest/databricks_ingestor.py#L158-L170】:

schema_info = ingestor.get_table_schema("sales.transactions")
for col in schema_info["columns"]:
    print(f"{col['name']}: {col['type']} (nullable: {col['nullable']})")

Table Lineage

Retrieve upstream and downstream dependencies via get_table_lineage()【/semantica/ingest/databricks_ingestor.py#L226-L236】:

lineage = ingestor.get_table_lineage(
    "main.analytics.revenue_report",
    include_column_lineage=False
)

print("Upstream tables:", lineage["upstream"])
print("Downstream tables:", lineage["downstream"])

---

Exporting to Semantica Document Format

The final step in connecting Databricks Unity Catalog to Semantica for KG ingestion is converting DatabricksData into Semantica documents. The export_as_documents() method【/semantica/ingest/databricks_ingestor.py#L78-L84】 maps table rows to graph-ready objects:

documents = ingestor.export_as_documents(
    data,
    id_field="customer_id",           # Unique identifier field

    text_fields=["name", "email", "company"]  # Fields to embed/index

)

# Feed to downstream KG components

for doc in documents:
    kg_pipeline.add_document(doc)

Each document preserves provenance metadata linking back to the source Unity Catalog table, essential for data lineage tracking in the knowledge graph.

---

Complete End-to-End Example

from semantica.ingest import DatabricksIngestor

# Initialize with environment variable fallback

with DatabricksIngestor() as ingestor:  # reads DATABRICKS_* env vars

    # Discover available tables

    tables = ingestor.list_tables(catalog="main", schema="sales")
    print(f"Available tables: {tables}")
    
    # Get schema info

    schema = ingestor.get_table_schema("main.sales.opportunities")
    
    # Ingest with limit

    raw_data = ingestor.ingest_table(
        "main.sales.opportunities",
        limit=50_000
    )
    
    # Convert to KG documents

    documents = ingestor.export_as_documents(
        raw_data,
        id_field="opportunity_id",
        text_fields=["name", "description", "account_name"]
    )
    
    print(f"Produced {len(documents)} documents for KG ingestion")

---

Error Handling and Observability

Every public method integrates with Semantica's progress tracker and logging infrastructure:

• self.progress_tracker.start_tracking() wraps operations with visual feedback【/semantica/ingest/databricks_ingestor.py#L61-L66】
• ProcessingError and ValidationError from semantica/utils/exceptions.py provide structured error propagation
• get_logger() from semantica/utils/logging.py ensures consistent log formatting

---

Summary

• Authentication: PAT or OAuth M2M via explicit parameters or DATABRICKS_* environment variables
• Core classes: DatabricksConnector for connections, DatabricksIngestor for operations, DatabricksData for results
• Ingestion methods: ingest_table() for full tables, ingest_query() for custom SQL
• Metadata access: list_catalogs(), list_schemas(), list_tables(), get_table_schema(), get_table_lineage()
• KG integration: export_as_documents() transforms rows to Semantica documents with configurable ID and text fields
• Resource safety: Context manager protocol ensures connection cleanup via __enter__/__exit__【/semantica/ingest/databricks_ingestor.py#L110-L117】

---

Frequently Asked Questions

What authentication methods does the Semantica Databricks connector support?

The connector supports personal access tokens (PAT) and OAuth M2M for service principals. In DatabricksConnector.connect(), PAT authentication uses the access_token parameter while OAuth M2M uses a credentials_provider callable【/semantica/ingest/databricks_ingestor.py#L68-L82】. Environment variables are automatically consulted if explicit parameters are omitted.

How does the ingestor handle large tables?

The ingest_table() method accepts a limit parameter for bounded extraction. For very large datasets, combine ingest_query() with predicate pushdown (filtering in SQL) rather than fetching entire tables. The DatabricksData container provides row_count for immediate visibility into result set size.

Can I retrieve column-level lineage from Unity Catalog?

Yes. Pass include_column_lineage=True to get_table_lineage()【/semantica/ingest/databricks_ingestor.py#L226-L236】. The underlying Databricks Lineage API returns detailed dependencies at the column granularity, though this may increase response latency for wide tables.

What happens to database-specific types during ingestion?

All values are automatically serialized to JSON-compatible Python types during extraction. The DatabricksData dataclass stores rows as dictionaries with native Python values, eliminating manual conversion of timestamps, decimals, or binary data before knowledge-graph ingestion.

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