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

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

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:

# 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 mirrors the Databricks API exactly, ensuring code portability.

Complete Snowflake Ingestion Example

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:

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

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	DatabricksConnector, DatabricksData, DatabricksIngestor implementations
semantica/ingest/snowflake_ingestor.py	SnowflakeConnector, SnowflakeData, SnowflakeIngestor implementations
semantica/ingest/__init__.py	Public API exports: DatabricksIngestor, SnowflakeIngestor
tests/test_databricks_ingestor.py	Unit tests for connection handling, pagination, error paths
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.

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.

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