# How to Ingest Data from Databricks and Snowflake into Knowledge Graphs with Semantica

> Easily ingest data from Databricks and Snowflake into knowledge graphs with Semantica. Our specialized ingestors streamline data extraction and conversion for graph construction.

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

---

**Semantica provides dedicated `DatabricksIngestor` and `SnowflakeIngestor` classes that extract data from cloud warehouses and convert it into documents ready for knowledge graph construction.**

To ingest data from Databricks and Snowflake into knowledge graphs, you use Semantica's unified ingestion API: create an ingestor instance, connect to your warehouse, pull tables or run custom queries, then export results as documents for downstream graph building. This guide covers the complete implementation based on the semantica-agi/semantica source code.

## Architecture of Semantica's Data Ingestion System

Semantica's ingestion layer follows a clean separation between **connection management** and **data extraction**. Understanding this architecture helps you troubleshoot issues and extend functionality.

### Core Components

- **Connector classes** (`DatabricksConnector`, `SnowflakeConnector`) — Handle authentication and low-level client creation in [`semantica/ingest/databricks_ingestor.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/ingest/databricks_ingestor.py) and [`semantica/ingest/snowflake_ingestor.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/ingest/snowflake_ingestor.py)

- **Data classes** (`DatabricksData`, `SnowflakeData`) — Store extracted rows, column names, and provenance metadata as dataclasses

- **Ingestor classes** (`DatabricksIngestor`, `SnowflakeIngestor`) — Provide the high-level API for table extraction and SQL queries

- **Document conversion** (`export_as_documents`) — Transforms tabular data into Semantica's standard document format for knowledge graph pipelines

Both ingestors implement an identical contract, making source swaps transparent to downstream code.

### Databricks Authentication Options

The `DatabricksConnector` class in [`semantica/ingest/databricks_ingestor.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/ingest/databricks_ingestor.py) supports two authentication flows:

- **Personal Access Token (PAT)** — Pass `token` directly to the constructor
- **OAuth M2M** — Service principal authentication for production workloads

Connection parameters include `host`, `http_path` (SQL warehouse endpoint), `catalog`, and `schema`. The connector lazily initializes Unity Catalog clients only when metadata methods are called.

### Snowflake Authentication Options

The `SnowflakeConnector` class in [`semantica/ingest/snowflake_ingestor.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/ingest/snowflake_ingestor.py) handles:

- **Username/password** authentication
- **Key-pair** authentication for enhanced security

Required parameters: `account`, `user`, `password` (or key pair), `role`, `warehouse`, `database`, and `schema`.

## Ingesting Data from Databricks into Knowledge Graphs

The `DatabricksIngestor` class provides three primary extraction methods: `ingest_table()` for full table reads, `ingest_query()` for custom SQL, and metadata discovery utilities.

### Complete Databricks Ingestion Example

```python
from semantica.ingest import DatabricksIngestor

# Initialize with your SQL warehouse credentials

ingestor = DatabricksIngestor(
    host="https://adb-01.azuredatabricks.net",
    token="dapi-XXXXXXXXXXXXXXXX",
    http_path="/sql/1.0/warehouses/0123abcd",
    catalog="main",
    schema="sales",
)

# Method 1: Ingest entire table with pagination

table_data = ingestor.ingest_table(
    table_name="customers",
    limit=5000,           # Cap total rows

    order_by="created_at", # Ensure deterministic pagination

    where_clause="country = 'US'"  # Optional filtering

)

# Method 2: Run parameterized custom query

query_data = ingestor.ingest_query(
    sql="SELECT * FROM orders WHERE order_date > :date AND status = :status",
    parameters={"date": "2024-01-01", "status": "shipped"}
)

# Convert to Semantica documents for knowledge graph construction

documents = ingestor.export_as_documents(
    data=table_data,
    id_field="customer_id",           # Optional: custom document ID

    text_fields=["first_name", "last_name", "notes"]  # Fields to concatenate as document text

)

```

The `export_as_documents()` method enriches each document with source metadata including `catalog`, `schema`, `table`, and `row_data`, enabling full lineage tracking in your knowledge graph.

### Databricks Metadata Discovery

Before ingestion, inspect available resources:

```python

# List catalogs and schemas

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

# Get table schema and lineage

columns = ingestor.get_table_schema("customers")
lineage = ingestor.get_table_lineage("orders")  # Unity Catalog lineage

```

## Ingesting Data from Snowflake into Knowledge Graphs

The `SnowflakeIngestor` class in [`semantica/ingest/snowflake_ingestor.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/ingest/snowflake_ingestor.py) mirrors the Databricks API exactly, ensuring code portability.

### Complete Snowflake Ingestion Example

```python
from semantica.ingest import SnowflakeIngestor

# Initialize with warehouse credentials

ingestor = SnowflakeIngestor(
    account="myaccount",
    user="etl_user",
    password="VerySecretPassword",
    role="ANALYST",
    warehouse="COMPUTE_WH",
    database="SALES_DB",
    schema="PUBLIC",
)

# Ingest table with automatic pagination for large datasets

products = ingestor.ingest_table("PRODUCTS", limit=10_000)

# Execute analytical query

reviews = ingestor.ingest_query(
    "SELECT id, description, rating FROM reviews WHERE rating >= 4 AND created_at > :cutoff",
    parameters={"cutoff": "2024-06-01"}
)

# Export to documents with configurable text aggregation

documents = ingestor.export_as_documents(
    data=reviews,
    text_fields=["description"],  # Primary content for embedding/ indexing

    id_field="id"
)

# documents ready for KGBuilder, RDFExporter, or vector store indexing

```

### Snowflake-Specific Features

The `SnowflakeIngestor` includes warehouse-specific optimizations:

- **Result set streaming** — Memory-efficient handling of multi-million row results
- **Session parameter injection** — Control Snowflake-specific behaviors via `session_parameters` dict
- **Warehouse auto-resume** — Detects and handles suspended warehouses

## Converting Ingested Data to Knowledge Graphs

The `export_as_documents()` method bridges raw data extraction and graph construction. Each document contains:

| Field | Source | Purpose in KG |
|-------|--------|---------------|
| `id` | Configurable column or UUID | Unique node identifier |
| `text` | Concatenated `text_fields` | Content for embedding, LLM reasoning, or full-text search |
| `metadata` | Source system provenance | Edge properties, temporal versioning, data lineage |
| `row_data` | Full record as dict | Structured properties for RDF/Property Graph conversion |

Feed documents to Semantica's knowledge graph components:

```python
from semantica.kg import KGBuilder

# Build graph from Databricks documents

builder = KGBuilder()
graph = builder.from_documents(
    documents,
    entity_extraction_model="gpt-4",
    relation_types=["purchased", "located_in", "reviewed"]
)

# Export to various graph formats

graph.to_rdf("output.ttl")      # RDF/OWL for semantic web tools

graph.to_neo4j(driver)          # Property graph to Neo4j

graph.to_amazon_neptune(session) # AWS Neptune-compatible format

```

## Progress Tracking and Observability

Both ingestors integrate Semantica's `ProgressTracker` and centralized logging via `get_logger()`.

```python
from semantica.utils.logging import get_logger

# Logs automatically capture connection events, query execution, and failures

logger = get_logger("ingestion")

# ProgressTracker reports granular step completion

# - Connection establishment

# - Query execution start/stop

# - Row batch retrieval (for pagination)

# - Document conversion

# - Final export

```

This instrumentation enables production monitoring and debugging without custom instrumentation code.

## Key Source Files and Test Coverage

| File | Purpose |
|------|---------|
| [`semantica/ingest/databricks_ingestor.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/ingest/databricks_ingestor.py) | `DatabricksConnector`, `DatabricksData`, `DatabricksIngestor` implementations |
| [`semantica/ingest/snowflake_ingestor.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/ingest/snowflake_ingestor.py) | `SnowflakeConnector`, `SnowflakeData`, `SnowflakeIngestor` implementations |
| [`semantica/ingest/__init__.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/ingest/__init__.py) | Public API exports: `DatabricksIngestor`, `SnowflakeIngestor` |
| [`tests/test_databricks_ingestor.py`](https://github.com/semantica-agi/semantica/blob/main/tests/test_databricks_ingestor.py) | Unit tests for connection handling, pagination, error paths |
| [`tests/test_snowflake_ingestor.py`](https://github.com/semantica-agi/semantica/blob/main/tests/test_snowflake_ingestor.py) | Unit tests for authentication, query execution, metadata |

The test suites validate authentication edge cases, large result set handling, and graceful degradation when warehouses are unavailable.

## Summary

- **Use `DatabricksIngestor`** for Unity Catalog and SQL warehouse data extraction with PAT or OAuth M2M authentication
- **Use `SnowflakeIngestor`** for warehouse data extraction with password or key-pair authentication
- **Both classes share identical APIs** — `ingest_table()`, `ingest_query()`, `export_as_documents()`, and metadata methods
- **Export documents feed directly** into `KGBuilder`, `RDFExporter`, or custom graph construction pipelines
- **Source metadata is preserved** in every document for complete knowledge graph lineage
- **Progress tracking and logging** are built-in for production observability

## Frequently Asked Questions

### What authentication methods does Semantica support for Databricks?

Semantica supports **Personal Access Token (PAT)** and **OAuth M2M (service principal)** authentication for Databricks. Pass the `token` parameter directly for PAT, or configure OAuth credentials via environment variables for M2M flows. The `DatabricksConnector` class handles token refresh and Unity Catalog client initialization automatically.

### Can I ingest from Snowflake using key-pair authentication instead of passwords?

Yes. The `SnowflakeConnector` class accepts `private_key` and `private_key_passphrase` parameters as alternatives to `password`. This is the recommended approach for production deployments, as implemented in [`semantica/ingest/snowflake_ingestor.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/ingest/snowflake_ingestor.py).

### How does `export_as_documents()` handle large tables with millions of rows?

Both ingestors implement **server-side pagination** transparently. The `ingest_table()` method automatically chunks large result sets using `limit` and `offset` mechanisms (Databricks) or result set iterators (Snowflake). The `export_as_documents()` method processes batches incrementally, keeping memory usage bounded regardless of source table size.

### Can I use the same downstream code for both Databricks and Snowflake sources?

Absolutely. The `DatabricksIngestor` and `SnowflakeIngestor` classes implement an identical interface. Both return data objects with the same methods and structure, and both use the same `export_as_documents()` signature. You can parameterize your pipeline by ingestor class without changing document-to-graph conversion logic.