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

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

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

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 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) 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

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

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:

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

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