# How to Ingest Data from Snowflake, SQL, and Databricks Sources into Semantica

> Easily ingest data from Snowflake, SQL, and Databricks into Semantica. Leverage dedicated ingestor classes to stream data into unified document formats for your knowledge-graph pipelines. Streamline your data integration today.

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

---

**Semantica provides dedicated `SnowflakeIngestor` and `DatabricksIngestor` classes that stream data from cloud warehouses into unified document formats for knowledge-graph pipelines.**

The `semantica-agi/semantica` repository ships with two production-ready ingestion modules. Both implement a **connector-ingestor-data class** architecture and support context-manager patterns, batched streaming, and automatic document export for downstream vector stores and knowledge-graph builders.

---

## Snowflake Data Ingestion

The Snowflake ingestion stack lives in [`semantica/ingest/snowflake_ingestor.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/ingest/snowflake_ingestor.py). It consists of three core components:

| Component | Class | Key Responsibilities |
|-----------|-------|----------------------|
| **Connector** | `SnowflakeConnector` | Manages `snowflake.connector` sessions, authentication (password, key-pair, OAuth, SSO), and connection lifecycle |
| **Ingestor** | `SnowflakeIngestor` | High-level API: `ingest_table()`, `ingest_query()`, schema introspection, document export |
| **Data Container** | `SnowflakeData` | Holds rows, columns, metadata, and ingestion timestamps |

### Snowflake Authentication Methods

`SnowflakeConnector` (line 76 in [`snowflake_ingestor.py`](https://github.com/semantica-agi/semantica/blob/main/snowflake_ingestor.py)) validates credentials from explicit arguments or environment variables:

- `SNOWFLAKE_ACCOUNT`
- `SNOWFLAKE_USER`
- `SNOWFLAKE_PASSWORD` (or private key path for key-pair auth)
- `SNOWFLAKE_WAREHOUSE`, `SNOWFLAKE_DATABASE`, `SNOWFLAKE_SCHEMA`

Missing required fields raise `ValidationError` from `semantica.utils.exceptions`.

### Ingesting a Snowflake Table

```python
from semantica.ingest import SnowflakeIngestor

ingestor = SnowflakeIngestor(
    account="myaccount",
    user="myuser",
    password="mypassword",
    warehouse="COMPUTE_WH",
    database="MYDB",
    schema="PUBLIC",
)

# Stream 5,000 rows with optional filtering and sorting

snow_data = ingestor.ingest_table(
    table_name="CUSTOMERS",
    limit=5000,
    where="COUNTRY = 'US'",
    order_by="CREATED_AT DESC",
)

# Export to Semantica document format

documents = ingestor.export_as_documents(
    snow_data,
    id_field="CUSTOMER_ID",
    text_fields=["NAME", "NOTES"],
)

```

`ingest_table()` (line 31) builds fully-qualified quoted identifiers (`"DB"."SCHEMA"."TABLE"`) and safely constructs `WHERE`, `ORDER BY`, `LIMIT`, and `OFFSET` clauses.

### Running Custom Snowflake Queries

```python
with SnowflakeIngestor(
    account="myaccount",
    user="myuser",
    password="mypassword",
) as ingestor:
    query_data = ingestor.ingest_query(
        "SELECT * FROM SALES WHERE SALE_DATE > %(date)s",
        params={"date": "2024-01-01"},
        batch_size=2000,  # stream 2k-row batches

    )
    print(f"Fetched {query_data.row_count} rows")

```

The context-manager pattern (`with` statement) guarantees connection teardown via `disconnect()`.

---

## Databricks Unity Catalog Ingestion

The Databricks module in [`semantica/ingest/databricks_ingestor.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/ingest/databricks_ingestor.py) adds Unity Catalog integration and lineage tracking on top of the same architectural pattern.

| Component | Class | Key Responsibilities |
|-----------|-------|----------------------|
| **Connector** | `DatabricksConnector` | Manages `databricks-sql-connector` sessions; supports PAT or OAuth client credentials |
| **Ingestor** | `DatabricksIngestor` | Table/query ingestion, Unity Catalog introspection, lineage retrieval |
| **Data Container** | `DatabricksData` | Stores rows with Databricks-specific metadata (catalog, schema, table) |

### Databricks Authentication

`DatabricksConnector` (line 79 in [`databricks_ingestor.py`](https://github.com/semantica-agi/semantica/blob/main/databricks_ingestor.py)) accepts:

- `host` (workspace URL) or `DATABRICKS_HOST` environment variable
- `token` (PAT) or `DATABRICKS_TOKEN`
- `http_path` to SQL warehouse
- OAuth: `client_id`, `client_secret` for service principal flows

OAuth credentials trigger `databricks.sdk.core.oauth_service_principal` token refresh.

### Ingesting Databricks Tables with Lineage

```python
from semantica.ingest import DatabricksIngestor

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

# Ingest with back-ticked identifiers (`catalog`.`schema`.`table`)

db_data = ingestor.ingest_table(
    table_name="customers",
    limit=10000,
    where="country = 'US'",
)

# Query Unity Catalog lineage

lineage = ingestor.get_table_lineage(
    table_name="customers",
    include_column_lineage=False,
)

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

```

`get_table_lineage()` calls the Unity Catalog lineage REST endpoint—unavailable in the Snowflake module.

### Databricks Document Export

```python
with DatabricksIngestor(
    host="https://adb-123.azuredatabricks.net",
    token="dapi-xxxxxxxxxxxxxxxx",
    http_path="/sql/1.0/warehouses/abcdef1234567890",
) as ingestor:
    docs = ingestor.export_as_documents(
        ingestor.ingest_table("orders", limit=5000),
        text_fields=["order_description", "notes"],
    )

```

The `source` metadata field is automatically set to `"databricks"` for downstream provenance tracking.

---

## Generic SQL Ingestion Pattern

While Semantica does not ship a dedicated "generic SQL" ingestor, both `SnowflakeIngestor` and `DatabricksIngestor` support arbitrary SQL via `ingest_query()`. For other databases (PostgreSQL, MySQL, BigQuery), implement the same connector-ingestor-data class pattern or use `ingest_query()` against compatible wire-protocol endpoints.

Key implementation requirements per the source:

1. **Connection management**: `connect()`, `disconnect()`, `test_connection()`
2. **Parameter binding**: Named placeholders with dict-based `params`
3. **Batched streaming**: Configurable `batch_size` to limit memory
4. **Document export**: `export_as_documents()` with `id_field` and `text_fields` selection

---

## Progress Tracking and Observability

Both ingestors integrate `semantica.utils.progress_tracker` to emit:

- Ingestion start events
- Intermediate batch progress
- Completion or failure status

Access via `ingestor.progress_tracker` or disable by passing `progress_tracker=None`.

---

## Error Handling

Custom exceptions from [`semantica/utils/exceptions.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/utils/exceptions.py):

- **`ValidationError`**: Missing credentials or malformed configuration
- **`ProcessingError`**: Query execution failures, connection timeouts, or export errors

Both ingestors wrap low-level driver exceptions and attach context (query, table name, batch number).

---

## Summary

- **Snowflake ingestion**: Use `SnowflakeIngestor` from [`semantica/ingest/snowflake_ingestor.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/ingest/snowflake_ingestor.py) with password, key-pair, or OAuth authentication; supports `ingest_table()`, `ingest_query()`, and INFORMATION_SCHEMA introspection.
- **Databricks ingestion**: Use `DatabricksIngestor` from [`semantica/ingest/databricks_ingestor.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/ingest/databricks_ingestor.py) with PAT or OAuth; adds Unity Catalog lineage (`get_table_lineage()`) and back-ticked identifier handling.
- **Document export**: Both classes provide `export_as_documents()` converting row batches to `{id, text, metadata}` dicts ready for vector stores or KG builders.
- **Always use context managers** (`with` statements) to ensure connection cleanup.

---

## Frequently Asked Questions

### What authentication methods does Semantica support for Snowflake ingestion?

`SnowflakeIngestor` supports password-based login, key-pair authentication (RSA private key), OAuth, and SSO via the underlying `snowflake-connector-python`. Credentials can be passed as constructor arguments or read from `SNOWFLAKE_*` environment variables. The connector validates required fields and raises `ValidationError` if any are missing.

### How does Databricks ingestion differ from Snowflake in Semantica?

`DatabricksIngestor` adds Unity Catalog integration: `list_catalogs()`, `list_schemas()`, and `get_table_lineage()` query Databricks' metadata and lineage REST endpoints. It uses back-ticked identifiers (`` `catalog`.`schema`.`table` ``) instead of Snowflake's double quotes. OAuth service principal flows are handled via `databricks.sdk.core.oauth_service_principal`.

### Can I ingest data from generic SQL databases not explicitly supported?

Semantica does not ship a generic SQL ingestor, but you can extend the same architecture: implement a connector class managing your driver's connection lifecycle, an ingestor class with `ingest_query()` and `ingest_table()` methods, and a data container. Alternatively, use database-specific foreign data wrappers or ODBC bridges that present as Snowflake-compatible endpoints.

### What is the recommended pattern for production ingestion jobs?

Use the context-manager pattern (`with DatabricksIngestor(...) as ingestor:`) to guarantee connection cleanup. Set explicit `batch_size` parameters to control memory usage when streaming large tables. Enable `progress_tracker` for observability in scheduled pipelines, and always validate credentials with `test_connection()` before expensive `ingest_table()` calls.