# How to Connect Databricks Unity Catalog to Semantica for KG Ingestion

> Easily connect Databricks Unity Catalog to Semantica for knowledge graph ingestion. Authenticate with PAT or OAuth M2M and ingest tables or queries efficiently.

- Repository: [Semantica /semantica](https://github.com/semantica-agi/semantica)
- Tags: how-to-guide
- Published: 2026-08-15

---

**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()`:

```python

# 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.

```python
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】:

```python
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】:

```python
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】:

```python

# 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】:

```python
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】:

```python
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:

```python
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

```python
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`](https://github.com/semantica-agi/semantica/blob/main/semantica/utils/exceptions.py) provide structured error propagation
- `get_logger()` from [`semantica/utils/logging.py`](https://github.com/semantica-agi/semantica/blob/main/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.