Forem: Reza Brianca

AI-Powered Intellectual Property Similarity Detection on Snowflake

Reza Brianca — Tue, 07 Apr 2026 23:10:00 +0000

Intellectual property offices worldwide face a growing challenge: how do you ensure that every new trademark, logo, or jingle is genuinely distinct from the millions already registered? For one of our prospect in public sector, this problem spans two fundamentally different media types: images (product logos, packaging designs) and audio (brand jingles, musical trademarks). Each new submission must be compared against the entire existing registry before it can be approved.

Manual review simply does not scale. An examiner can visually compare a handful of logos per hour, but the backlog grows faster than human reviewers can process it. Audio comparison is even harder. You can imagine listening to thousands of jingles to catch a subtle similarity is impractical. Subjectivity compounds the problem: two examiners may disagree on whether a logo is "too similar" to an existing registration. And consistency matters: the same pair of assets should always produce the same similarity verdict, regardless of which examiner handles the case or what time of day it is reviewed.

The traditional approach by hiring more examiners and building manual review queues is not practical since it will introduce higher cost, latency, and inconsistency. What if you could reduce the comparison to a mathematical operation? Convert each IP asset into a numerical fingerprint (a vector embedding), store those fingerprints in a database, and compute similarity scores automatically.

This article walks through how we built exactly that: an AI-powered IP similarity detection system that runs entirely within Snowflake, combining GPU-accelerated deep learning models with native vector search and a Streamlit interface for IP examiners.

Solution Architecture

The system is built on four Snowflake capabilities working together:

Snowpark Container Services (SPCS) for GPU-backed model inference
The native VECTOR data type for embedding storage
VECTOR_COSINE_SIMILARITY for similarity search
Streamlit in Snowflake for the examiner-facing UI

The data flow is straightforward: raw files sit on a Snowflake internal stage. SPCS containers mount that stage directly as a volume, so the Flask-based inference services read files from /stage without any data transfer. Service function uses SQL UDFs that wrap HTTP calls to the containers. This makes embedding generation callable from any SQL query. The resulting vectors are stored in registry tables and compared via VECTOR_COSINE_SIMILARITY.

A critical design choice was keeping everything within Snowflake's governance boundary. For a public-sector organizations, data sovereignty is non-negotiable. No IP assets leave the platform. This means the models run inside SPCS, the files live on Snowflake stages, and the embeddings are stored in Snowflake tables. Access control is handled through Snowflake RBAC, and every query is captured in Snowflake's access history for audit purposes.

This architecture also means there is no external API dependency during inference. The models are loaded into containers at startup and serve requests entirely within the Snowflake network. If internet access is revoked after deployment, the system continues to function normally. This is an important consideration for government networks with strict egress policies.

Model Selection: Why the Right Embedding Matters

Choosing the right embedding model is the most consequential architectural decision in a similarity search system. The model determines what "similar" means. For IP registration system, getting this wrong renders the entire system useless.

Image Pipeline: Vision Transformer (ViT)

For image similarity, we use Google's vit-base-patch16-224-in21k, a Vision Transformer pretrained on ImageNet-21k. It produces 768-dimensional embeddings extracted from the CLS token of the final hidden layer. ViT is a strong general-purpose choice: it captures both low-level visual features (color, shape, texture) and higher-level semantic patterns (object composition, layout). No fine-tuning was needed for our use case since the pretrained model already distinguishes product packaging designs effectively.

from transformers import AutoFeatureExtractor, AutoModel
import torch

model_id = "google/vit-base-patch16-224-in21k"
extractor = AutoFeatureExtractor.from_pretrained(model_id)
model = AutoModel.from_pretrained(model_id)

inputs = extractor(images=image, return_tensors="pt")
with torch.no_grad():
    outputs = model(**inputs)
embedding = outputs.last_hidden_state[:, 0]  # 768-dim CLS token

Audio Pipeline: PANNs CNN14

Audio is where the model selection gets interesting. The natural first choice might be CLAP (laion/larger_clap_music), a contrastive language-audio model that has become popular for audio understanding tasks. However, during prototyping we discovered a fundamental problem: CLAP produces semantic embeddings that cluster audio by genre and type. When we tested it against our jingle corpus, all files within a similar genre scored 0.93-0.99 cosine similarity. A pop jingle and a completely different pop jingle were nearly indistinguishable. This makes CLAP excellent for genre classification but useless for IP distinctiveness detection.

PANNs CNN14 (Pretrained Audio Neural Networks) solves this problem. Trained on AudioSet's 527 sound classes across roughly 5,000 hours of audio, CNN14 produces 2048-dimensional acoustic embeddings that capture the unique spectral characteristics of each file including timbre, rhythm, frequency patterns, and temporal structure. Two pop jingles that sound superficially similar will produce meaningfully different embeddings because PANNs encodes their actual acoustic fingerprint rather than their semantic category.

from panns_inference import AudioTagging
import librosa
import numpy as np

at = AudioTagging(checkpoint_path=None, device="cpu")
audio_array, _ = librosa.load("jingle.mp3", sr=32000, mono=True)
audio_array = audio_array[np.newaxis, :]
_, embedding = at.inference(audio_array)  # 2048-dim acoustic embedding

This distinction between acoustic and semantic embeddings is critical. For IP registration, you need to answer "does this specific audio file sound like that specific audio file?" and not "are these two files in the same genre?"

Implementation Highlights

Deploying Models as SPCS Services

Each model runs in its own Docker container, deployed as an SPCS service with GPU resources. The Docker images are built locally, pushed to Snowflake's built-in image registry, and referenced in the service specification. The containers mount the Snowflake stage as a volume at /stage, enabling direct file access. The Flask app simply reads from the filesystem without any data serialization or network transfer overhead.

Here is the abbreviated service creation for the image pipeline:

CREATE SERVICE IMAGE_EMBEDDING_SERVICE
  IN COMPUTE POOL POOL_GPU_S
  FROM SPECIFICATION $$
  spec:
    containers:
    - name: image-embedding
      image: /img_demo/raw/img_repo/image-embedding-service:latest
      resources:
        requests:
          nvidia.com/gpu: 1
      volumeMounts:
      - name: stage-volume
        mountPath: /stage
    endpoints:
    - name: embed
      port: 8080
      public: false
    volumes:
    - name: stage-volume
      source: "@IMG_DEMO.RAW.IP_STAGE"
  $$
  MIN_INSTANCES = 1
  MAX_INSTANCES = 1;

The readinessProbe on the /health endpoint ensures Snowflake only routes traffic to containers that have fully loaded their models. ViT takes a few seconds, while PANNs CNN14 needs to download its ~300 MB checkpoint on first run.

The key pattern here is the service function. This is an SQL UDF that routes calls to the container's HTTP endpoint. This abstraction lets any SQL query generate embeddings without knowing anything about Docker, Flask, or PyTorch:

CREATE FUNCTION GENERATE_IMAGE_EMBEDDING(FILE_PATH VARCHAR)
  RETURNS VARIANT
  SERVICE = IMAGE_EMBEDDING_SERVICE
  ENDPOINT = embed
  AS '/embed-file';

Similarity Search in SQL

With embeddings stored as native VECTOR columns, similarity search becomes a single SQL query. The pattern uses a CTE to generate the candidate embedding, then cross-joins against the registry with VECTOR_COSINE_SIMILARITY:

WITH uploaded_embedding AS (
    SELECT GENERATE_IMAGE_EMBEDDING('uploads/new_logo.png')::VECTOR(FLOAT, 768) AS emb
)
SELECT
    r.IP_ID,
    r.FILE_PATH,
    VECTOR_COSINE_SIMILARITY(r.IMAGE_EMBEDDING, u.emb) AS SIMILARITY
FROM IMAGE_IP_REGISTRY r
CROSS JOIN uploaded_embedding u
WHERE r.STATUS = 'ACTIVE'
ORDER BY SIMILARITY DESC;

This query generates the embedding for the new submission, compares it against every active registration, and returns results ranked by similarity. All of this can be done in a single SQL statement. The ::VECTOR(FLOAT, 768) cast converts the VARIANT output from the service function into Snowflake's native VECTOR type, which is required for VECTOR_COSINE_SIMILARITY to operate.

The same pattern applies to audio, substituting GENERATE_AUDIO_EMBEDDING and VECTOR(FLOAT, 2048) for the 2048-dimensional PANNs embeddings.

Threshold-Based Classification

Similarity scores are classified into four tiers that drive the examiner's workflow:

Similarity Score	Category	Action
> 0.95	DUPLICATE	Reject registration
0.85 -- 0.95	HIGHLY SIMILAR	Manual review required
0.70 -- 0.85	MODERATE	Flag for review
< 0.70	DISTINCT	Safe to register

These thresholds were calibrated using sample data from prospect's existing registry and should be adjusted based on real-world feedback during production use. The boundaries are not hard-coded in SQL. They live in the application layer, making them easy to tune without redeploying any infrastructure.

Streamlit UI: The Examiner's Workflow

The Streamlit in Snowflake application provides a tabbed interface for image and audio similarity detection. The examiner's workflow is:

Upload - drag and drop an image or audio file into the uploader
Compare - the system uploads to stage, generates an embedding via the service function, and runs VECTOR_COSINE_SIMILARITY against all active registry entries
Review - a color-coded verdict badge (red/orange/yellow/green) appears alongside a table of all comparison scores and a preview of the closest match
Register - if the verdict is DISTINCT, a registration form appears where the examiner enters metadata (brand, product, title, genre) and submits to insert the new record

The app also includes a gallery view showing all currently registered IP assets with presigned URLs for previewing images and playing audio directly in the browser. Service readiness is checked on page load, with a warning banner displayed if either embedding service is still warming up.

Operational Considerations

Cost Optimization

GPU compute pools are not required to run 24/7. Since public sectors operate during standard business hours, we implemented scheduled Snowflake Tasks to resume services at 07:45 and suspend them at 17:30 (Jakarta time), giving 15 minutes for GPU nodes to provision and containers to pass readiness probes before the 8 AM workday begins:

CREATE TASK RESUME_SERVICES_MORNING
  SCHEDULE = 'USING CRON 45 7 * * MON-FRI Asia/Bangkok'
AS
BEGIN
  ALTER COMPUTE POOL POOL_GPU_S RESUME;
  ALTER SERVICE IMAGE_EMBEDDING_SERVICE RESUME;
  ALTER SERVICE AUDIO_EMBEDDING_SERVICE RESUME;
END;

The Streamlit app's compute pool (POOL_CPU_S) uses AUTO_RESUME and AUTO_SUSPEND_SECS instead, waking automatically when an examiner visits the URL and suspending after 5 minutes of inactivity. This two-tier approach to explicitly schedule for GPU pools and automatic management for lightweight CPU pools will balance cost control with user experience.

Health Monitoring

A stored procedure (CHECK_HEALTH) runs 10 minutes after the morning resume, checking compute pool and service status and logging results to a SPCS_HEALTH_LOG table. This provides operational visibility without requiring manual checks.

Governance and Access Control

The system uses a minimal-privilege EXAMINER_ROLE for examiners, granting only USAGE on the database, schema, warehouse, and Streamlit app. The Streamlit app executes with owner's rights, so examiners never need direct access to the registry tables, service functions, or compute pools. Combined with Snowflake's built-in access history logging and encryption at rest and in transit, this satisfies the audit and compliance requirements typical of public-sector deployments.

Scaling for Large Registries

For initial data migration from legacy systems, the batch embedding generation processes files in chunks using LIMIT/OFFSET patterns to avoid query timeouts. A seed manifest table tracks which files have been processed, allowing the migration to be resumed if interrupted. The SPCS compute pool supports auto-scaling (MIN_NODES to MAX_NODES) to handle concurrent users during peak registration periods.

For very large registries (100,000+ assets), the current brute-force VECTOR_COSINE_SIMILARITY scan across all rows will eventually need to be supplemented with approximate nearest neighbor (ANN) indexing. Snowflake's vector search capabilities continue to evolve in this area, and the registry table schema is already structured to take advantage of future optimizations without any migration.

Results and Takeaways

Building this system validated several key points:

Pretrained models eliminate the cold-start problem.
Using HuggingFace models (ViT and PANNs CNN14) that are already trained on large-scale datasets meant we could go from concept to working prototype without any model training. The total model weight is roughly 650 MB combined, which is a trivial size for a GPU container.

Model selection is the critical decision.
The CLAP vs. PANNs comparison demonstrated that a seemingly reasonable model choice (CLAP is newer, more popular, and supports text-audio queries) can be fundamentally wrong for the use case. Acoustic embeddings and semantic embeddings answer different questions and IP similarity detection requires acoustic fidelity.

Snowflake SPCS makes model deployment operationally simple.
The pattern of Docker container + SPCS service + service function turns a PyTorch model into a SQL-callable function. There is no separate inference infrastructure to manage, no API gateway to configure, and no data movement between systems.

The solution extends naturally.
The same architecture supports internal use via Streamlit or external access via API endpoints. Additional media types (video, 3D models) can be added by deploying new embedding services and registry tables following the same pattern.

For public-sector organizations managing intellectual property at scale, this approach offers a path to automated, consistent, and auditable similarity detection. This can all be done within a single governed platform using Snowflake.

Building a True Lakehouse via Snowflake Postgres and pg_lake

Reza Brianca — Wed, 04 Mar 2026 05:05:00 +0000

Overview

Recently, Snowflake acquired Crunchy Data, which provides an enterprise-grade PostgreSQL offering. On top of that awesome news, the team also developed and decided to publish pg_lake, a PostgreSQL extension to manage Iceberg tables. This extension provides write and read capability into Iceberg tables via PostgreSQL.

Given the trend on the open data format, I was curious. If PostgreSQL can write into the Iceberg table, then the true lakehouse architecture can be achieved in a scalable manner. In this article, I want to explain how we can use Snowflake Postgres (recently GA) and Snowflake to manage the entire data lakehouse architecture.
Here is the high level architecture on the lakehouse we will build

Architecture and Value Proposition

Here are the components involved in this architecture:

Snowflake Postgres for OLTP writes
pg_lake as the datalake components
AWS S3 as the storage layer
Snowflake as the powerful data and AI platform

By developing this architecture, we hope to achieve the following:

True openness using Iceberg format
Transactional consistency and schema evolution supported by pg_lake
Workload separation between OLTP (on Postgres) and OLAP (on Snowflake)
Leverage zero-copy architecture

Prerequisites

Create a Snowflake account where the Snowflake Postgres is available. You can check the region availability here.
Create an AWS account since we will use AWS S3 and its IAM services.
Local client applications like dBeaver to run the PostgreSQL command.

Setup in Snowflake Postgres

We can easily spin up the Postgres instance using SQL. By default, the instance can not be accessed so we need to add the NETWORK_RULE and NETWORK_POLICY to make it accessible from the client endpoint. We will create 2 types of table here: normal Postgres table and another Iceberg table. Get the metadata file path to be registered when creating a table in Snowflake later. Please note the PostgreSQL instance creation may take about 5-10 minutes. Here is the SQL we can use to set this up:

CREATE POSTGRES INSTANCE retail_txns
  COMPUTE_FAMILY = 'STANDARD_M'
  STORAGE_SIZE_GB = 50
  AUTHENTICATION_AUTHORITY = POSTGRES
  COMMENT = 'Retail simulation with pg_lake Iceberg integration';
-- This will create 2 users : snowflake_admin and application with the password
-- Copy the password since it will only appear once

-- Get the connection host (use this in psql/dBeaver to connect)
SHOW POSTGRES INSTANCES;

-- Check the instance status, wait until `state` changes to `READY`
DESCRIBE POSTGRES INSTANCE retail_txns;

-- The network setup will need a database to attach into
CREATE DATABASE IF NOT EXISTS RETAIL_ANALYTICS;
CREATE SCHEMA IF NOT EXISTS RETAIL_ANALYTICS.NETWORK;

CREATE OR REPLACE NETWORK RULE RETAIL_ANALYTICS.NETWORK.PG_RETAIL_INGRESS
  TYPE = IPV4
  VALUE_LIST = ('0.0.0.0/0') -- need to change this for production environment
  MODE = POSTGRES_INGRESS;

CREATE OR REPLACE NETWORK POLICY PG_RETAIL_ALLOW_ALL
  ALLOWED_NETWORK_RULE_LIST = ('RETAIL_ANALYTICS.NETWORK.PG_RETAIL_INGRESS')
  COMMENT = 'Allow connections to retail_txns Postgres instance';

-- Attach the network policy to the instance
-- (changes may take up to 2 minutes to take effect)
ALTER POSTGRES INSTANCE retail_txns
  SET NETWORK_POLICY = 'PG_RETAIL_ALLOW_ALL';

You should now be able to connect to your PostgreSQL instance using psql or tools like dBeaver. The successful creation will look like this in the Snowsight.

Attach Storage Integration to Postgres

Once the instance is ready, we need to create and attach a STORAGE_INTEGRATION to inform Postgres that it can write and read the Iceberg table into the specified S3-compatible bucket. The type is called POSTGRES_EXTERNAL_STORAGE. We need to create the role in the AWS. After login to AWS Console, go to (or search for) IAM (Identity and Access Management) menu. We will need to create a new role to register this storage to Postgres.

We need to go to the AWS Console to create the Policy which then will be attached into the Role. Go to IAM → Policies. You can also follow the documentation here.

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": [
        "s3:PutObject",
        "s3:GetObject",
        "s3:GetObjectVersion",
        "s3:DeleteObject",
        "s3:DeleteObjectVersion"
      ],
      "Resource": "arn:aws:s3:::retail-lake-ap3/*"
    },
    {
      "Effect": "Allow",
      "Action": [
        "s3:ListBucket",
        "s3:GetBucketLocation"
      ],
      "Resource": "arn:aws:s3:::retail-lake-ap3"
    }
  ]
}

Next, go to IAM → Roles and create the new role and attach the Policy we just created. Note the role ARN value since we will use this for the Storage Integration setup.

IMPORTANT: Set the IAM role's "Maximum session duration" to 12 hours (default 1 hour). The storage integration will NOT work with the default 1-hour session.

-- This integration type (POSTGRES_EXTERNAL_STORAGE) allows the Postgres
-- instance to write Iceberg data to your own S3 bucket via pg_lake. 

CREATE STORAGE INTEGRATION pg_lake_s3_integration
  TYPE = POSTGRES_EXTERNAL_STORAGE
  STORAGE_PROVIDER = 'S3'
  ENABLED = TRUE
  STORAGE_AWS_ROLE_ARN = 'arn:aws:iam::012345678910:role/snowflake-retail-lake-role'
  STORAGE_ALLOWED_LOCATIONS = ('s3://retail-lake-ap3/');

-- Retrieve the Snowflake IAM user ARN and external ID
DESCRIBE STORAGE INTEGRATION pg_lake_s3_integration;
-- Record these values from the output:
--   STORAGE_AWS_IAM_USER_ARN  (e.g., arn:aws:iam::<account>:user/snowflake-postgres-integration-management)
--   STORAGE_AWS_EXTERNAL_ID   (e.g., ABC12345_SFCRole=1_abcdefg)

Go back to the AWS Console and access the IAM → Policy to edit the configuration.

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Principal": {
        "AWS": "<STORAGE_AWS_IAM_USER_ARN from DESCRIBE output>"
      },
      "Action": "sts:AssumeRole",
      "Condition": {
        "StringEquals": {
          "sts:ExternalId": "<STORAGE_AWS_EXTERNAL_ID from DESCRIBE output>"
        }
      }
    }
  ]
}

Finally, we can attach this storage integration into the Snowflake Postgres. This will allow Postgres instances to write and read into the S3 bucket we specified above.

ALTER POSTGRES INSTANCE retail_txns
  SET STORAGE_INTEGRATION = pg_lake_s3_integration;

Insert Initial Data in Postgres

We can then connect to the Postgres instance using psql or dBeaver to set up the initial data and Iceberg tables.

psql "host=<host_from_SHOW_POSTGRES_INSTANCE_above> port=5432 dbname=postgres user=snowflake_admin sslmode=require"

Once we are inside the instance, we can run the following commands:

-- Create pg_lake extension (installs all sub-extensions)
CREATE EXTENSION IF NOT EXISTS pg_lake CASCADE;

-- Set the default Iceberg storage location (persistent across all sessions)
ALTER DATABASE postgres
  SET pg_lake_iceberg.default_location_prefix = 's3://retail-lake-ap3/';

-- Apply to current session immediately
SET pg_lake_iceberg.default_location_prefix TO 's3://retail-lake-ap3/';

-- Verify
SHOW pg_lake_iceberg.default_location_prefix;
-- Expected: s3://retail-lake-ap3/frompg/tables/

We can put some initial data in the database using the familiar DDL.

-- Create retail OLTP tables (standard heap tables for fast inserts) 

-- Stores
CREATE TABLE IF NOT EXISTS stores (
    store_id    SERIAL PRIMARY KEY,
    store_name  TEXT NOT NULL,
    city        TEXT NOT NULL,
    state       TEXT NOT NULL,
    opened_at   TIMESTAMP DEFAULT now()
);

-- Products
CREATE TABLE IF NOT EXISTS products (
    product_id    SERIAL PRIMARY KEY,
    product_name  TEXT NOT NULL,
    category      TEXT NOT NULL,
    unit_price    NUMERIC(10,2) NOT NULL,
    created_at    TIMESTAMP DEFAULT now()
);

-- Customers
CREATE TABLE IF NOT EXISTS customers (
    customer_id   SERIAL PRIMARY KEY,
    first_name    TEXT NOT NULL,
    last_name     TEXT NOT NULL,
    email         TEXT UNIQUE NOT NULL,
    loyalty_tier  TEXT DEFAULT 'BRONZE',
    created_at    TIMESTAMP DEFAULT now()
);

-- Transactions (header)
CREATE TABLE IF NOT EXISTS transactions (
    txn_id        SERIAL PRIMARY KEY,
    store_id      INT NOT NULL REFERENCES stores(store_id),
    customer_id   INT REFERENCES customers(customer_id),
    txn_timestamp TIMESTAMP NOT NULL DEFAULT now(),
    total_amount  NUMERIC(12,2) NOT NULL DEFAULT 0,
    payment_method TEXT NOT NULL DEFAULT 'CARD',
    synced_at     TIMESTAMP  -- NULL means not yet synced to Iceberg
);

-- Transaction line items
CREATE TABLE IF NOT EXISTS transaction_items (
    item_id       SERIAL PRIMARY KEY,
    txn_id        INT NOT NULL REFERENCES transactions(txn_id),
    product_id    INT NOT NULL REFERENCES products(product_id),
    quantity      INT NOT NULL DEFAULT 1,
    unit_price    NUMERIC(10,2) NOT NULL,
    line_total    NUMERIC(12,2) NOT NULL,
    synced_at     TIMESTAMP
);

-- Indexes for efficient sync queries (only scan unsynced rows)
CREATE INDEX IF NOT EXISTS idx_txn_synced ON transactions(synced_at) WHERE synced_at IS NULL;
CREATE INDEX IF NOT EXISTS idx_items_synced ON transaction_items(synced_at) WHERE synced_at IS NULL;

-- Seed reference data 

INSERT INTO stores (store_name, city, state) VALUES
    ('Downtown Flagship', 'Seattle', 'WA'),
    ('Mall of Pacific', 'San Francisco', 'CA'),
    ('Eastside Express', 'Bellevue', 'WA'),
    ('Harbor Point', 'Portland', 'OR'),
    ('Valley Center', 'San Jose', 'CA');

INSERT INTO products (product_name, category, unit_price) VALUES
    ('Organic Coffee Beans 1lb',   'Grocery',     14.99),
    ('Wireless Earbuds Pro',       'Electronics', 79.99),
    ('Cotton T-Shirt',             'Apparel',     24.99),
    ('Stainless Water Bottle',     'Home',        19.99),
    ('Running Shoes',              'Footwear',    129.99),
    ('Yoga Mat Premium',           'Fitness',     39.99),
    ('LED Desk Lamp',              'Electronics', 44.99),
    ('Organic Protein Bar 12pk',   'Grocery',     29.99),
    ('Backpack 30L',               'Accessories', 59.99),
    ('Sunscreen SPF50',            'Personal Care', 12.99),
    ('Bluetooth Speaker',          'Electronics', 49.99),
    ('Trail Mix 2lb',              'Grocery',      9.99),
    ('Insulated Lunch Bag',        'Home',        22.99),
    ('Fitness Tracker Band',       'Electronics', 89.99),
    ('Bamboo Cutting Board',       'Home',        17.99);

INSERT INTO customers (first_name, last_name, email, loyalty_tier) VALUES
    ('Alice',   'Chen',     'alice.chen@example.com',    'GOLD'),
    ('Bob',     'Martinez', 'bob.martinez@example.com',  'SILVER'),
    ('Carol',   'Johnson',  'carol.j@example.com',       'BRONZE'),
    ('David',   'Kim',      'david.kim@example.com',     'GOLD'),
    ('Eve',     'Patel',    'eve.patel@example.com',     'PLATINUM'),
    ('Frank',   'Lopez',    'frank.lopez@example.com',   'BRONZE'),
    ('Grace',   'Wang',     'grace.wang@example.com',    'SILVER'),
    ('Hank',    'Brown',    'hank.brown@example.com',    'BRONZE'),
    ('Iris',    'Davis',    'iris.davis@example.com',    'GOLD'),
    ('Jack',    'Wilson',   'jack.wilson@example.com',   'SILVER');

We will then continue to create the Iceberg table. Since we already set the storage integration, creating an Iceberg table is straightforward.

CREATE TABLE iceberg_transactions (
    txn_id         INT,
    store_id       INT,
    customer_id    INT,
    txn_timestamp  TIMESTAMP,
    total_amount   NUMERIC(12,2),
    payment_method TEXT,
    synced_at      TIMESTAMP
) USING iceberg;

CREATE TABLE iceberg_transaction_items (
    item_id      INT,
    txn_id       INT,
    product_id   INT,
    quantity     INT,
    unit_price   NUMERIC(10,2),
    line_total   NUMERIC(12,2),
    synced_at    TIMESTAMP
) USING iceberg;

-- Dimension tables (Iceberg copies for fully self-contained analytics on Snowflake)
CREATE TABLE iceberg_stores (
    store_id    INT,
    store_name  TEXT,
    city        TEXT,
    state       TEXT,
    opened_at   TIMESTAMP
) USING iceberg;

CREATE TABLE iceberg_products (
    product_id    INT,
    product_name  TEXT,
    category      TEXT,
    unit_price    NUMERIC(10,2),
    created_at    TIMESTAMP
) USING iceberg;

CREATE TABLE iceberg_customers (
    customer_id   INT,
    first_name    TEXT,
    last_name     TEXT,
    email         TEXT,
    loyalty_tier  TEXT,
    created_at    TIMESTAMP
) USING iceberg;

-- Seed dimension data into Iceberg (one-time)
INSERT INTO iceberg_stores    SELECT * FROM stores;
INSERT INTO iceberg_products  SELECT * FROM products;
INSERT INTO iceberg_customers SELECT * FROM customers;

At this point, we will have 10 tables with 5 base tables and 5 iceberg tables (it’s called foreign tables in PostgreSQL).

Once we create the Iceberg tables inside the Postgres, we will need to get the metadata path.

SELECT table_name, metadata_location FROM iceberg_tables;

-- Example metadata_location:
--   s3://retail-lake-ap3/frompg/tables/postgres/public/iceberg_transactions/
--     <OID>/metadata/00000-xxxx.metadata.json

The data will look like this

Setup in Snowflake

The setup on the Snowflake side has 2 parts. One is to register the metadata path since the open format like Iceberg uses this to capture the latest information of the tables. The second is to register the data path itself.

Attach External Volume

Setup different external volumes to enable us to read the data. This is a critical component to support open-format tables. We will use the task in the next section to read the latest metadata periodically. This can be improved to run automatically if you use other data catalogue tools. We can use this command to set this up.

CREATE OR REPLACE EXTERNAL VOLUME retail_lake_volume
  STORAGE_LOCATIONS = (
    (
      NAME                    = 'retail-s3-ap3'
      STORAGE_PROVIDER        = 'S3'
      STORAGE_BASE_URL        = 's3://retail-lake-ap3/'
      STORAGE_AWS_ROLE_ARN    = 'arn:aws:iam::012345678910:role/snowflake-retail-lake-role'
    )
  );

-- Retrieve the Snowflake IAM user ARN and external ID
DESCRIBE EXTERNAL VOLUME retail_lake_volume;
-- Copy the STORAGE_AWS_IAM_USER_ARN and STORAGE_AWS_EXTERNAL_ID from the output

We will need to update the IAM Policy again.

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Principal": {
        "AWS": [
          "<STORAGE_AWS_IAM_USER_ARN from DESCRIBE STORAGE INTEGRATION pg_lake_s3_integration>",
          "<STORAGE_AWS_IAM_USER_ARN from DESCRIBE EXTERNAL VOLUME retail_lake_volume>"
        ]
      },
      "Action": "sts:AssumeRole",
      "Condition": {
        "StringEquals": {
          "sts:ExternalId": [
            "<STORAGE_AWS_EXTERNAL_ID from DESCRIBE STORAGE INTEGRATION pg_lake_s3_integration>",
            "<STORAGE_AWS_EXTERNAL_ID from DESCRIBE EXTERNAL VOLUME retail_lake_volume>"
          ]
        }
      }
    }
  ]
}

Once we save this, we can check to verify the connection.

-- Verify connectivity
SELECT SYSTEM$VERIFY_EXTERNAL_VOLUME('retail_lake_volume');

Attach Catalog and Metadata Path for Iceberg Data

Setup external volume and catalog integration to enable Snowflake to read the actual parquet data in the S3-compatible bucket. We will need to put the metadata file path created during the Postgres setup.

CREATE DATABASE IF NOT EXISTS RETAIL_ANALYTICS;
CREATE SCHEMA IF NOT EXISTS RETAIL_ANALYTICS.LAKEHOUSE;
USE DATABASE RETAIL_ANALYTICS;
USE SCHEMA RETAIL_ANALYTICS.LAKEHOUSE;

CREATE OR REPLACE CATALOG INTEGRATION pg_lake_catalog
  CATALOG_SOURCE = OBJECT_STORE
  TABLE_FORMAT = ICEBERG
  ENABLED = TRUE;

Since pg_lake manages the Iceberg metadata (not Snowflake), we use a catalog integration for object storage files, and point each table to its metadata file using METADATA_FILE_PATH. This path is relative to the external volume's STORAGE_BASE_URL. Given STORAGE_BASE_URL = 's3://retail-lake-ap3/', and a metadata_location will look like this
s3://retail-lake-ap3/frompg/tables/postgres/public/iceberg_transactions/<OID>/metadata/00000-xxxx.metadata.json. The METADATA_FILE_PATH would be
frompg/tables/postgres/public/iceberg_transactions/<OID>/metadata/00000-xxxx.metadata.json
We will then create the Iceberg table in Snowflake using the metadata file path for each table.

-- Transactions fact table
-- Replace METADATA_FILE_PATH with actual path from: SELECT metadata_location FROM iceberg_tables WHERE table_name = 'iceberg_transactions';
CREATE OR REPLACE ICEBERG TABLE iceberg_transactions
  EXTERNAL_VOLUME = 'retail_lake_volume'
  CATALOG = 'pg_lake_catalog'
  METADATA_FILE_PATH = 'frompg/tables/postgres/public/iceberg_transactions/<OID>/metadata/00000-xxxx.metadata.json';

-- Repeat for another tables with different iceberg table name

At this point, the Iceberg tables created in the Postgres instance are already accessible in Snowflake. We can check this in the Database Explorer to confirm.

However, any update in the Iceberg tables will not be reflected since we don’t have the mechanism to refresh the metadata. Therefore, we will need to create a task to check for the update periodically.

-- ---- Storage integration for stages ----
CREATE OR REPLACE STORAGE INTEGRATION iceberg_stage_s3_integration
  TYPE = EXTERNAL_STAGE
  STORAGE_PROVIDER = 'S3'
  ENABLED = TRUE
  STORAGE_AWS_ROLE_ARN = 'arn:aws:iam::01234578910:role/snowflake-retail-lake-role'
  STORAGE_ALLOWED_LOCATIONS = ('s3://retail-lake-ap3/frompg/tables/');

-- Retrieve the IAM user ARN and external ID for this integration.
-- Add these to your IAM role trust policy
DESCRIBE STORAGE INTEGRATION iceberg_stage_s3_integration;

We will need to update the IAM Policy again for the last time. This is the final look of the IAM Policy for this architecture.

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Principal": {
        "AWS": [
          "<STORAGE_AWS_IAM_USER_ARN from DESCRIBE STORAGE INTEGRATION pg_lake_s3_integration>",
          "<STORAGE_AWS_IAM_USER_ARN from DESCRIBE EXTERNAL VOLUME retail_lake_volume>",
          "<STORAGE_AWS_IAM_USER_ARN from DESCRIBE STORAGE INTEGRATION iceberg_stage_s3_integration>"
        ]
      },
      "Action": "sts:AssumeRole",
      "Condition": {
        "StringEquals": {
          "sts:ExternalId": [
            "<STORAGE_AWS_EXTERNAL_ID from DESCRIBE STORAGE INTEGRATION pg_lake_s3_integration>",
            "<STORAGE_AWS_EXTERNAL_ID from DESCRIBE EXTERNAL VOLUME retail_lake_volume>",
            "<STORAGE_AWS_EXTERNAL_ID from DESCRIBE STORAGE INTEGRATION iceberg_stage_s3_integration>"
          ]
        }
      }
    }
  ]
}

We will create the EXTERNAL STAGE pointing to the metadata folder in S3

-- Directory stages (one per Iceberg table's metadata directory  
CREATE OR REPLACE STAGE stg_meta_transactions
  URL = 's3://retail-lake-ap3/frompg/tables/postgres/public/iceberg_transactions/<OID>/metadata/'
  STORAGE_INTEGRATION = iceberg_stage_s3_integration
  DIRECTORY = (ENABLE = TRUE);
-- Repeat for another tables using different STAGE name

There will be 5 stages created in total

To enable metadata changes in the stage, we can use the combination of Task, Stream and Stored Procedure in Snowflake.

-- Create stream to monitor changes
CREATE OR REPLACE STREAM stream_meta_transactions ON STAGE stg_meta_transactions;
-- repeat for another stage 

-- Create Procedure to refresh the iceberg tables
CREATE OR REPLACE PROCEDURE sp_refresh_iceberg_transactions()
RETURNS VARCHAR
LANGUAGE SQL
AS
$$
DECLARE
  latest_path VARCHAR;
BEGIN
  SELECT RELATIVE_PATH INTO :latest_path
  FROM DIRECTORY(@stg_meta_transactions)
  WHERE RELATIVE_PATH LIKE '%.metadata.json'
  ORDER BY LAST_MODIFIED DESC
  LIMIT 1;

  EXECUTE IMMEDIATE
    'ALTER ICEBERG TABLE iceberg_transactions REFRESH '''
    || 'frompg/tables/postgres/public/iceberg_transactions/<OID>/metadata/' || latest_path || '''';
  RETURN latest_path;
END;
$$;
-- repeat for another iceberg tables

-- Create task to do this periodically
CREATE OR REPLACE TASK task_refresh_stage_transactions
  WAREHOUSE = COMPUTE_WH
  SCHEDULE  = '2 MINUTES'
AS
  ALTER STAGE stg_meta_transactions REFRESH;

CREATE OR REPLACE TASK task_refresh_iceberg_transactions
  WAREHOUSE = COMPUTE_WH
  AFTER task_refresh_stage_transactions
  WHEN SYSTEM$STREAM_HAS_DATA('stream_meta_transactions')
AS
  CALL sp_refresh_iceberg_transactions();
-- create the same task pattern for other tables

-- Start the task
ALTER TASK task_refresh_iceberg_transactions RESUME;
ALTER TASK task_refresh_stage_transactions RESUME;
-- Stop the task
-- ALTER TASK task_refresh_iceberg_transactions SUSPEND;
-- ALTER TASK task_refresh_stage_transactions SUSPEND;

-- We can also trigger this manually
ALTER STAGE stg_meta_transactions REFRESH;
CALL sp_refresh_iceberg_transactions();

All of these will be available via the Database Explorer

Simulate the Transaction

Feel free to use or modify the sample python code below.

import argparse
import random
import time
import sys
from datetime import datetime

import psycopg2
from psycopg2.extras import execute_values


# Retail simulation constants

PAYMENT_METHODS = ["CARD", "CASH", "DIGITAL_WALLET", "GIFT_CARD"]
ITEMS_PER_TXN_RANGE = (1, 5)  # each transaction has 1-5 line items


def get_reference_data(conn):
    """Load store, product, and customer IDs from the database."""
    with conn.cursor() as cur:
        cur.execute("SELECT store_id FROM stores")
        store_ids = [r[0] for r in cur.fetchall()]

        cur.execute("SELECT product_id, unit_price FROM products")
        products = [(r[0], float(r[1])) for r in cur.fetchall()]

        cur.execute("SELECT customer_id FROM customers")
        customer_ids = [r[0] for r in cur.fetchall()]

    return store_ids, products, customer_ids


def generate_transaction(store_ids, products, customer_ids):
    """Generate a single random retail transaction with line items."""
    store_id = random.choice(store_ids)
    # 90% of transactions have a known customer, 10% are anonymous
    customer_id = random.choice(customer_ids) if random.random() < 0.9 else None
    payment = random.choice(PAYMENT_METHODS)
    num_items = random.randint(*ITEMS_PER_TXN_RANGE)

    items = []
    total = 0.0
    selected_products = random.sample(products, min(num_items, len(products)))

    for product_id, unit_price in selected_products:
        qty = random.randint(1, 3)
        line_total = round(unit_price * qty, 2)
        total += line_total
        items.append((product_id, qty, unit_price, line_total))

    return store_id, customer_id, round(total, 2), payment, items


def insert_batch(conn, batch):
    """Insert a batch of transactions and their line items using bulk operations."""
    with conn.cursor() as cur:
        # Bulk-insert all transactions in one round-trip
        txn_rows = [
            (store_id, customer_id, total, payment)
            for store_id, customer_id, total, payment, _items in batch
        ]
        result = execute_values(
            cur,
            """
            INSERT INTO transactions (store_id, customer_id, total_amount, payment_method)
            VALUES %s
            RETURNING txn_id
            """,
            txn_rows,
            fetch=True,
        )
        txn_ids = [r[0] for r in result]

        # Collect all line items across every transaction
        all_items = []
        for txn_id, (_sid, _cid, _tot, _pay, items) in zip(txn_ids, batch):
            for pid, qty, up, lt in items:
                all_items.append((txn_id, pid, qty, up, lt))

        # Bulk-insert all line items in one round-trip
        if all_items:
            execute_values(
                cur,
                """
                INSERT INTO transaction_items (txn_id, product_id, quantity, unit_price, line_total)
                VALUES %s
                """,
                all_items,
            )
    conn.commit()


def trigger_sync(conn):
    """Manually trigger the Iceberg sync (CTE chains INSERT then UPDATE atomically)."""
    with conn.cursor() as cur:
        cur.execute(
            "WITH synced AS ("
            "INSERT INTO iceberg_transactions "
            "SELECT txn_id, store_id, customer_id, txn_timestamp, "
            "total_amount, payment_method, now() "
            "FROM transactions WHERE synced_at IS NULL "
            "RETURNING txn_id) "
            "UPDATE transactions SET synced_at = now() "
            "WHERE txn_id IN (SELECT txn_id FROM synced)"
        )
        cur.execute(
            "WITH synced AS ("
            "INSERT INTO iceberg_transaction_items "
            "SELECT item_id, txn_id, product_id, quantity, "
            "unit_price, line_total, now() "
            "FROM transaction_items WHERE synced_at IS NULL "
            "RETURNING item_id) "
            "UPDATE transaction_items SET synced_at = now() "
            "WHERE item_id IN (SELECT item_id FROM synced)"
        )
    conn.commit()


def print_stats(conn):
    """Print current counts from both heap and Iceberg tables."""
    with conn.cursor() as cur:
        cur.execute("SELECT COUNT(*) FROM transactions")
        heap_txn = cur.fetchone()[0]
        cur.execute("SELECT COUNT(*) FROM transactions WHERE synced_at IS NOT NULL")
        synced_txn = cur.fetchone()[0]
        cur.execute("SELECT COUNT(*) FROM iceberg_transactions")
        ice_txn = cur.fetchone()[0]
        cur.execute("SELECT COUNT(*) FROM transaction_items")
        heap_items = cur.fetchone()[0]
        cur.execute("SELECT COUNT(*) FROM iceberg_transaction_items")
        ice_items = cur.fetchone()[0]

    print(f"  Heap transactions: {heap_txn} (synced: {synced_txn})")
    print(f"  Iceberg transactions: {ice_txn}")
    print(f"  Heap line items: {heap_items}")
    print(f"  Iceberg line items: {ice_items}")


def main():
    parser = argparse.ArgumentParser(description="Transaction simulator")
    parser.add_argument("--host", required=True, help="Postgres host ")
    parser.add_argument("--port", type=int, default=5432)
    parser.add_argument("--dbname", default="postgres")
    parser.add_argument("--user", default="snowflake_admin")
    parser.add_argument("--password", required=True, help="Postgres admin password")
    parser.add_argument(
        "--transactions",
        type=int,
        default=5000,
        help="Total number of transactions to generate",
    )
    parser.add_argument(
        "--batch-size",
        type=int,
        default=500,
        help="Number of transactions per commit batch",
    )
    parser.add_argument(
        "--sync-every",
        type=int,
        default=500,
        help="Trigger manual Iceberg sync after this many transactions (0 = rely on pg_cron only)",
    )
    args = parser.parse_args()

    conn_params = dict(
        host=args.host,
        port=args.port,
        dbname=args.dbname,
        user=args.user,
        password=args.password,
        sslmode="require",
    )

    conn = psycopg2.connect(**conn_params)
    conn.autocommit = False

    print(f"Connected to {args.host}:{args.port}/{args.dbname}")
    store_ids, products, customer_ids = get_reference_data(conn)
    print(
        f"Loaded reference data: {len(store_ids)} stores, "
        f"{len(products)} products, {len(customer_ids)} customers"
    )

    total_generated = 0
    start_time = time.time()

    print(f"\nGenerating {args.transactions} transactions (batch size: {args.batch_size})...\n")

    while total_generated < args.transactions:
        batch_count = min(args.batch_size, args.transactions - total_generated)
        batch = [
            generate_transaction(store_ids, products, customer_ids)
            for _ in range(batch_count)
        ]

        insert_batch(conn, batch)
        total_generated += batch_count

        elapsed = time.time() - start_time
        tps = total_generated / elapsed if elapsed > 0 else 0
        print(
            f"  [{datetime.now().strftime('%H:%M:%S')}] "
            f"{total_generated}/{args.transactions} transactions "
            f"({tps:.0f} txn/sec)"
        )

        # Trigger manual sync at intervals
        if args.sync_every > 0 and total_generated % args.sync_every == 0:
            print(f"  >> Triggering Iceberg sync at {total_generated} transactions...")
            sync_start = time.time()
            trigger_sync(conn)
            print(f"  >> Sync completed in {time.time() - sync_start:.2f}s")

    # Final sync
    print("\nTriggering final Iceberg sync...")
    sync_start = time.time()
    trigger_sync(conn)
    print(f"Final sync completed in {time.time() - sync_start:.2f}s")

    elapsed = time.time() - start_time
    print(f"\n{'='*60}")
    print(f"Simulation complete!")
    print(f"  Total transactions: {total_generated}")
    print(f"  Elapsed time: {elapsed:.1f}s")
    print(f"  Average throughput: {total_generated / elapsed:.0f} txn/sec")
    print(f"{'='*60}")

    print("\nFinal counts:")
    print_stats(conn)

    print("\nIceberg metadata locations:")
    with conn.cursor() as cur:
        cur.execute("SELECT table_name, metadata_location FROM iceberg_tables")
        for row in cur.fetchall():
            print(f"  {row[0]}: {row[1]}")

    conn.close()
    print("\nDone. Data is now available on both Postgres and Snowflake via shared Iceberg tables.")


if __name__ == "__main__":
    main()

The script above will do the following 2 tasks

Insert transaction data to Postgres This is the typical INSERT INTO operation into your Postgres database.
Trigger periodic data sync to write into Iceberg tables We will need to run a periodic INSERT INTO for the Iceberg tables. Since this table is using columnar-format like parquet, it is much more efficient to do bulk insert rather than individual insert like we do for OLTP. I did a simulation to do bulk insert into Iceberg tables for every 500-1000 rows in Postgres and it can write in under 2 seconds consistently.

Trigger metadata and Iceberg table refresh

At this point, Snowflake does not know if the data is recently updated. That's why we need to get the latest metadata information along with the latest Iceberg table information. This is covered in the latest SQL script.

Once the sync is complete, we can run some queries to check if the data is flowing as expected.

-- 1. Check row counts match Postgres Iceberg tables
SELECT 'sf_transactions' AS source, COUNT(*) AS row_count FROM iceberg_transactions
UNION ALL
SELECT 'sf_items',        COUNT(*) FROM iceberg_transaction_items
UNION ALL
SELECT 'sf_stores',       COUNT(*) FROM iceberg_stores
UNION ALL
SELECT 'sf_products',     COUNT(*) FROM iceberg_products
UNION ALL
SELECT 'sf_customers',    COUNT(*) FROM iceberg_customers;

-- 2. Revenue by store (proves joins work across Iceberg tables)
SELECT
    s.store_name,
    COUNT(DISTINCT t.txn_id)  AS txn_count,
    SUM(t.total_amount)       AS revenue
FROM iceberg_transactions t
JOIN iceberg_stores s ON t.store_id = s.store_id
GROUP BY s.store_name
ORDER BY revenue DESC;

Closing

Combining Snowflake Postgres with S3-compatible bucket truly unlocks the lakehouse architecture. Snowflake separates the billing between Snowflake Postgres and the Virtual Warehouse used for analytics. You can check in more details here. We can improve this further by having a data catalogue tool so Snowflake can activate the AUTO_REFRESH attributes in the stage. Hope this helps and feel free to reach out if you have any questions!

How to Ingest Your Legacy Database to Snowflake

Reza Brianca — Wed, 21 Jan 2026 03:00:00 +0000

Overview

I recently came across a discussion with potential customers about their ingestion pain points. While they want to have a near real-time ingestion experience, they are currently using PostgreSQL version 9.2 where it’s already out of support since 8 years ago (2017). It was interesting because given the lack of IT resources (it was a small team), the database version upgrade will be a stretch for them. This is also on top of another challenge to convince management since the result of software upgrade is (most likely) intangible. Currently they have more than 15 TB with more than 200k transactions per day. We were asked on how to do the ingestion without having to upgrade their systems along with the cost consciousness mindset.

In this article, I want to share what was the approach I decided to take given those limitations. This approach may not be the perfect solution (and I welcome to hear more if you have more efficient solutions), but I think this can be relevant for most enterprises since I believe many of them are running a legacy system, the budget is minimal (if any), and there is no direct impact to revenue when the IT team wants to perform an improvement in the systems.

Solution Design

Since the current PostgreSQL version is not supported with the recent CDC tools like Debezium (the stable version from Debezium requires PostgreSQL version 12 and above), we need to create additional tables inside the database to capture these changes. These tables then need to be registered to Kafka in order to enable Snowpipe Streaming. The high level diagram looks like this:

Setup Procedure to Capture Table Changes

First, we need to create a table to capture the changes we want to input later. This is the SQL pseudocode:

CREATE TABLE transaction_cdc_log (
    cdc_id SERIAL PRIMARY KEY,
    table_name VARCHAR(100) NOT NULL,
    operation VARCHAR(10) NOT NULL CHECK (operation IN ('INSERT', 'UPDATE', 'DELETE')),
    record_id INTEGER NOT NULL,
    old_data TEXT,
    new_data TEXT,
    change_timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    change_user VARCHAR(100),
    processed BOOLEAN DEFAULT false
);

We can modify the column as needed. The idea is to capture the current and new data for any changes (INSERT, UPDATE or DELETE operation) in the table.

Next, we need to create an intermediary function to convert the old and new data into a full joined text as follows:

CREATE OR REPLACE FUNCTION transaction_row_to_text(rec transactions) RETURNS TEXT AS $$
BEGIN
    RETURN '{' ||
        '"transaction_id":' || COALESCE(rec.transaction_id::TEXT, 'null') || ',' ||
        '"transaction_code":"' || COALESCE(rec.transaction_code, '') || '",' ||
        '"outlet_id":' || COALESCE(rec.outlet_id::TEXT, 'null') || ',' ||
        '"transaction_date":"' || COALESCE(rec.transaction_date::TEXT, '') || '",' ||
        '"total_amount":' || COALESCE(rec.total_amount::TEXT, 'null') || ',' ||
        '"total_discount":' || COALESCE(rec.total_discount::TEXT, 'null') || ',' ||
        '"payment_method":"' || COALESCE(rec.payment_method, '') || '",' ||
        '"cashier_name":"' || COALESCE(rec.cashier_name, '') || '",' ||
        '"customer_id":"' || COALESCE(rec.customer_id, '') || '",' ||
        '"status":"' || COALESCE(rec.status, '') || '",' ||
        '"created_at":"' || COALESCE(rec.created_at::TEXT, '') || '"' ||
        '}';
END;
$$ LANGUAGE plpgsql;

Setup Trigger to Each Table

Once the function is ready, the next step is to create a function that returns a TRIGGER in the database. This function then will be called whenever any changes are made to the table we want to monitor.

CREATE OR REPLACE FUNCTION capture_transaction_changes() RETURNS TRIGGER AS $$
BEGIN
    IF (TG_OP = 'DELETE') THEN
        INSERT INTO transaction_cdc_log (table_name, operation, record_id, old_data, new_data, change_user)
        VALUES (TG_TABLE_NAME, TG_OP, OLD.transaction_id, transaction_row_to_text(OLD), NULL, current_user);
        RETURN OLD;
    ELSIF (TG_OP = 'UPDATE') THEN
        INSERT INTO transaction_cdc_log (table_name, operation, record_id, old_data, new_data, change_user)
        VALUES (TG_TABLE_NAME, TG_OP, NEW.transaction_id, transaction_row_to_text(OLD), transaction_row_to_text(NEW), current_user);
        RETURN NEW;
    ELSIF (TG_OP = 'INSERT') THEN
        INSERT INTO transaction_cdc_log (table_name, operation, record_id, old_data, new_data, change_user)
        VALUES (TG_TABLE_NAME, TG_OP, NEW.transaction_id, NULL, transaction_row_to_text(NEW), current_user);
        RETURN NEW;
    END IF;
    RETURN NULL;
END;
$$ LANGUAGE plpgsql;

The function above returns a TRIGGER which then can be used to monitor the desired table like the following:

CREATE TRIGGER transaction_cdc_trigger
AFTER INSERT OR UPDATE OR DELETE ON transactions -- this is the table name
FOR EACH ROW EXECUTE PROCEDURE capture_transaction_changes();

Up until this point, you may realize that we need to create those functions for each table we want to monitor. As mentioned in the Overview section, this is an approach we can use if there is no option to upgrade the database version. Of course I recommend upgrading so these processes can be handled seamlessly using more appropriate tools.

Setup Kafka Connect to CDC Log.

Once the function is ready, we can use Kafka Connect with JDBC Source Connectors to perform periodic pooling and publish this as Kafka Topic. The high level configuration file is as follows:

{
  "name": "pos-cdc-source",
  "config": {
    "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
    "tasks.max": "2",
    "connection.url": "jdbc:postgresql://pos-db:5432/pos_db",
    "connection.user": "your_username",
    "connection.password": "your_password",
    "mode": "timestamp+incrementing",
    "incrementing.column.name": "cdc_id",
    "timestamp.column.name": "change_timestamp",
    "table.whitelist": "transaction_cdc_log",
    "topic.prefix": "pos-",
    "poll.interval.ms": "1000",
    "batch.max.rows": "100",
    "timestamp.delay.interval.ms": "0",
    "validate.non.null": "false",
    "transforms": "createKey,extractInt",
    "transforms.createKey.type": "org.apache.kafka.connect.transforms.ValueToKey",
    "transforms.createKey.fields": "cdc_id",
    "transforms.extractInt.type": "org.apache.kafka.connect.transforms.ExtractField$Key",
    "transforms.extractInt.field": "cdc_id"
  }
}

Setup Snowpipe Streaming

Once the data changes are published via Kafka, we can configure Snowpipe Streaming to ingest the data into Snowflake. Please note that to handle ingestion efficiently, we need to use the ‘batch’ approach rather than insert. The batch period will depend on the configuration, so we can set this in seconds (the config above used 1 second). The high level python script is as follows:

#!/usr/bin/env python3
"""
Snowpipe Streaming Client with Batch Processing
Consumes CDC events from Kafka and streams them to Snowflake in batches
"""
import os
import sys
import json
import time
import logging
import signal
from datetime import datetime
from collections import defaultdict
from typing import Optional, Callable
from kafka import KafkaConsumer
import snowflake.connector
from dotenv import load_dotenv

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    handlers=[logging.StreamHandler(sys.stdout)]
)
logger = logging.getLogger(__name__)

# Load environment variables
load_dotenv()

class SnowpipeStreamingClient:
    def __init__(self):
        # Kafka configuration
        self.kafka_bootstrap_servers = os.getenv('KAFKA_BOOTSTRAP_SERVERS', 'kafka:9092')
        self.kafka_group_id = os.getenv('KAFKA_GROUP_ID', 'snowpipe-consumer-group')

        # Snowflake configuration
        self.snowflake_account = os.getenv('SNOWFLAKE_ACCOUNT')
        self.snowflake_user = os.getenv('SNOWFLAKE_USER')
        self.snowflake_password = os.getenv('SNOWFLAKE_PASSWORD')
        self.snowflake_database = os.getenv('SNOWFLAKE_DATABASE', 'DB_DEMO')
        self.snowflake_warehouse = os.getenv('SNOWFLAKE_WAREHOUSE', 'COMPUTE_WH')
        self.snowflake_role = os.getenv('SNOWFLAKE_ROLE', 'ACCOUNTADMIN')

        # Batch configuration
        self.batch_size = int(os.getenv('BATCH_SIZE', '50'))
        self.batch_timeout = float(os.getenv('BATCH_TIMEOUT', '2.0'))  # seconds

        # Validate Snowflake credentials
        if not all([self.snowflake_account, self.snowflake_user, self.snowflake_password]):
            raise ValueError("Missing Snowflake credentials. Please set SNOWFLAKE_ACCOUNT, SNOWFLAKE_USER, and SNOWFLAKE_PASSWORD environment variables.")

        # Initialize connections
        self.snowflake_conn = None
        self.consumer = None

        # Batch buffers by topic
        self.batches = {
            'pos-transaction_cdc_log': []
        }
        self.last_flush_time = time.time()

        # Stats
        self.records_processed = 0
        self.batches_flushed = 0
        self.errors = 0
        self.last_report_time = time.time()

        self._shutdown_requested = False
        self._setup_signal_handlers()

        logger.info(f"Snowpipe Streaming Client initialized (Batch Size: {self.batch_size}, Timeout: {self.batch_timeout}s)")

    def _setup_signal_handlers(self) -> None:
        signal.signal(signal.SIGINT, self._signal_handler)
        signal.signal(signal.SIGTERM, self._signal_handler)

    def _signal_handler(self, signum: int, frame) -> None:
        logger.info(f"Received signal {signum}, initiating graceful shutdown...")
        self._shutdown_requested = True

    def _deduplicate_batch(self, records: list, key_field: str) -> list:
        """Deduplicate records by key, keeping only the latest (last) occurrence"""
        seen = {}
        for idx, record in enumerate(records):
            new_data = self.parse_json_field(record.get('new_data'))
            if new_data and record.get('operation') != 'DELETE':
                key = new_data.get(key_field)
                if key is not None:
                    seen[key] = (idx, record)

        return [record for idx, record in sorted(seen.values(), key=lambda x: x[0])]

    def connect_snowflake(self):
        """Connect to Snowflake"""
        max_retries = 5
        for i in range(max_retries):
            try:
                self.snowflake_conn = snowflake.connector.connect(
                    account=self.snowflake_account,
                    user=self.snowflake_user,
                    password=self.snowflake_password,
                    database=self.snowflake_database,
                    warehouse=self.snowflake_warehouse,
                    role=self.snowflake_role
                )
                logger.info(f"Connected to Snowflake: {self.snowflake_account}")
                return
            except Exception as e:
                if i < max_retries - 1:
                    logger.warning(f"Snowflake connection attempt {i+1} failed: {e}")
                    logger.info("Retrying in 10 seconds...")
                    time.sleep(10)
                else:
                    raise

    def connect_kafka(self):
        """Connect to Kafka and subscribe to topics"""
        topics = list(self.batches.keys())

        max_retries = 10
        for i in range(max_retries):
            try:
                self.consumer = KafkaConsumer(
                    *topics,
                    bootstrap_servers=self.kafka_bootstrap_servers,
                    group_id=self.kafka_group_id,
                    auto_offset_reset='earliest',
                    enable_auto_commit=True,
                    auto_commit_interval_ms=1000,
                    max_poll_records=500,
                    max_poll_interval_ms=300000,
                    session_timeout_ms=30000,
                    heartbeat_interval_ms=10000,
                    fetch_min_bytes=1,
                    fetch_max_wait_ms=500,
                    value_deserializer=lambda m: json.loads(m.decode('utf-8'))
                )
                logger.info(f"Connected to Kafka: {self.kafka_bootstrap_servers}")
                logger.info(f"Subscribed to topics: {', '.join(topics)}")
                return
            except Exception as e:
                if i < max_retries - 1:
                    logger.warning(f"Kafka connection attempt {i+1} failed: {e}")
                    logger.info("Retrying in 10 seconds...")
                    time.sleep(10)
                else:
                    raise

    def parse_json_field(self, json_str):
        """Parse JSON-like string from PostgreSQL trigger"""
        if not json_str:
            return None
        try:
            return json.loads(json_str)
        except json.JSONDecodeError:
            return json_str

    def flush_transaction_batch(self, records: list) -> None:
        """Flush a batch of transaction records to Snowflake"""
        if not records:
            return

        deduped_records = self._deduplicate_batch(records, 'transaction_id')
        if not deduped_records:
            return

        try:
            cursor = self.snowflake_conn.cursor()

            values_list = []
            params = []

            for record in deduped_records:
                new_data = self.parse_json_field(record.get('new_data'))

                values_list.append("(%s, %s, %s, %s::TIMESTAMP_NTZ, %s, %s, %s, %s, %s, %s, %s::TIMESTAMP_NTZ)")
                params.extend([
                    new_data.get('transaction_id'),
                    new_data.get('transaction_code'),
                    new_data.get('outlet_id'),
                    new_data.get('transaction_date'),
                    new_data.get('total_amount'),
                    new_data.get('total_discount'),
                    new_data.get('payment_method'),
                    new_data.get('cashier_name'),
                    new_data.get('customer_id'),
                    new_data.get('status'),
                    new_data.get('created_at')
                ])

            if not values_list:
                cursor.close()
                return

            values_clause = ",\n".join(values_list)

            merge_sql = f"""
                MERGE INTO {self.snowflake_database}.POS_DATA.transactions t
                USING (
                    SELECT 
                        column1 as transaction_id,
                        column2 as transaction_code,
                        column3 as outlet_id,
                        column4 as transaction_date,
                        column5 as total_amount,
                        column6 as total_discount,
                        column7 as payment_method,
                        column8 as cashier_name,
                        column9 as customer_id,
                        column10 as status,
                        column11 as created_at
                    FROM VALUES {values_clause}
                ) s
                ON t.transaction_id = s.transaction_id
                WHEN MATCHED THEN UPDATE SET
                    transaction_code = s.transaction_code,
                    outlet_id = s.outlet_id,
                    transaction_date = s.transaction_date,
                    total_amount = s.total_amount,
                    total_discount = s.total_discount,
                    payment_method = s.payment_method,
                    cashier_name = s.cashier_name,
                    customer_id = s.customer_id,
                    status = s.status
                WHEN NOT MATCHED THEN INSERT (
                    transaction_id, transaction_code, outlet_id, transaction_date,
                    total_amount, total_discount, payment_method, cashier_name,
                    customer_id, status, created_at
                ) VALUES (
                    s.transaction_id, s.transaction_code, s.outlet_id, s.transaction_date,
                    s.total_amount, s.total_discount, s.payment_method, s.cashier_name,
                    s.customer_id, s.status, s.created_at
                )
            """

            cursor.execute(merge_sql, params)
            cursor.close()
            self.batches_flushed += 1

        except Exception as e:
            logger.error(f"Error flushing transaction batch ({len(deduped_records)} records): {e}")
            self.errors += len(deduped_records)

    def flush_batches(self, force=False):
        """Flush all batches to Snowflake if they meet size/timeout thresholds"""
        current_time = time.time()
        time_since_last_flush = current_time - self.last_flush_time

        for topic, batch in self.batches.items():
            should_flush = force or len(batch) >= self.batch_size or time_since_last_flush >= self.batch_timeout

            if should_flush and batch:
                if topic == 'pos-transaction_cdc_log':
                    self.flush_transaction_batch(batch)

                self.records_processed += len(batch)
                self.batches[topic].clear()

        if force or time_since_last_flush >= self.batch_timeout:
            self.last_flush_time = current_time

    def process_message(self, message):
        """Add message to appropriate batch buffer"""
        try:
            topic = message.topic
            kafka_value = message.value

            # Extract payload from Kafka Connect wrapper
            if isinstance(kafka_value, dict) and 'payload' in kafka_value:
                record = kafka_value['payload']
            else:
                record = kafka_value

            if not record or not isinstance(record, dict):
                logger.warning(f"Invalid record structure: {record}")
                self.errors += 1
                return

            # Add to batch buffer
            if topic in self.batches:
                self.batches[topic].append(record)
            else:
                logger.warning(f"Unknown topic: {topic}")
                return

            # Flush if any batch reaches size threshold
            self.flush_batches(force=False)

        except Exception as e:
            logger.error(f"Error processing message: {e}")
            self.errors += 1

    def run(self):
        """Main loop to consume and process messages"""
        logger.info("=" * 80)
        logger.info("Snowpipe Streaming Client Started (Batch Mode)")
        logger.info(f"Kafka: {self.kafka_bootstrap_servers}")
        logger.info(f"Snowflake: {self.snowflake_account}/{self.snowflake_database}")
        logger.info(f"Batch Size: {self.batch_size} | Timeout: {self.batch_timeout}s")
        logger.info("=" * 80)

        logger.info("Waiting 30 seconds for Kafka Connect to be ready...")
        time.sleep(30)

        # Connect
        self.connect_snowflake()
        self.connect_kafka()

        logger.info("Ready to process CDC events in batch mode")

        try:
            while not self._shutdown_requested:
                messages = self.consumer.poll(timeout_ms=1000, max_records=self.batch_size)

                for topic_partition, records in messages.items():
                    for message in records:
                        self.process_message(message)

                self.flush_batches(force=False)

                if time.time() - self.last_report_time >= 10:
                    total_buffered = sum(len(b) for b in self.batches.values())
                    logger.info(f"Processed: {self.records_processed} | Batches: {self.batches_flushed} | "
                               f"Buffered: {total_buffered} | Errors: {self.errors}")
                    self.last_report_time = time.time()

        except Exception as e:
            logger.error(f"Fatal error: {e}")
        finally:
            logger.info("Shutting down streaming client...")
            self.flush_batches(force=True)
        self._cleanup()

    def _cleanup(self) -> None:
        if self.consumer:
            self.consumer.close()
        if self.snowflake_conn:
            self.snowflake_conn.close()

        logger.info("=" * 80)
        logger.info("Streaming Client Stopped")
        logger.info(f"Total Records Processed: {self.records_processed}")
        logger.info(f"Total Batches Flushed: {self.batches_flushed}")
        logger.info(f"Total Errors: {self.errors}")
        logger.info("=" * 80)

if __name__ == '__main__':
    try:
        client = SnowpipeStreamingClient()
        client.run()
    except Exception as e:
        logger.error(f"Failed to start streaming client: {e}")
        sys.exit(1)

Create Docker Compose

To deploy this in a development environment easily, we can use Docker. This is the sample docker compose file we can use (assuming the update or new transactions are done separately).

services:
  # Zookeeper
  zookeeper:
    image: confluentinc/cp-zookeeper:7.5.0
    container_name: zookeeper
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000
    ports:
      - "2181:2181"
    networks:
      - pharma-network
    healthcheck:
      test: ["CMD", "nc", "-z", "localhost", "2181"]
      interval: 10s
      timeout: 5s
      retries: 5

  # Kafka Broker
  kafka:
    image: confluentinc/cp-kafka:7.5.0
    container_name: kafka
    depends_on:
      zookeeper:
        condition: service_healthy
    ports:
      - "9092:9092"
      - "9093:9093"
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092,PLAINTEXT_HOST://localhost:9093
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_AUTO_CREATE_TOPICS_ENABLE: "true"
    networks:
      - pos-network
    healthcheck:
      test: ["CMD", "kafka-broker-api-versions", "--bootstrap-server", "localhost:9092"]
      interval: 10s
      timeout: 10s
      retries: 5

  # Kafka Connect
  kafka-connect:
    image: confluentinc/cp-kafka-connect:7.5.0
    container_name: kafka-connect
    depends_on:
      kafka:
        condition: service_healthy
      pos-db:
        condition: service_started
      promo-db:
        condition: service_started
    ports:
      - "8083:8083"
    environment:
      CONNECT_BOOTSTRAP_SERVERS: kafka:9092
      CONNECT_REST_PORT: 8083
      CONNECT_GROUP_ID: "connect-group"
      CONNECT_CONFIG_STORAGE_TOPIC: "connect-configs"
      CONNECT_OFFSET_STORAGE_TOPIC: "connect-offsets"
      CONNECT_STATUS_STORAGE_TOPIC: "connect-status"
      CONNECT_CONFIG_STORAGE_REPLICATION_FACTOR: 1
      CONNECT_OFFSET_STORAGE_REPLICATION_FACTOR: 1
      CONNECT_STATUS_STORAGE_REPLICATION_FACTOR: 1
      CONNECT_KEY_CONVERTER: "org.apache.kafka.connect.json.JsonConverter"
      CONNECT_VALUE_CONVERTER: "org.apache.kafka.connect.json.JsonConverter"
      CONNECT_REST_ADVERTISED_HOST_NAME: "kafka-connect"
      CONNECT_PLUGIN_PATH: "/usr/share/java,/usr/share/confluent-hub-components"
    volumes:
      - ./kafka-connect/connectors:/etc/kafka-connect/connectors
    networks:
      - pos-network
    command:
      - bash
      - -c
      - |
        echo "Installing JDBC Connector..."
        confluent-hub install --no-prompt confluentinc/kafka-connect-jdbc:10.7.4
        echo "Downloading PostgreSQL JDBC Driver..."
        cd /usr/share/confluent-hub-components/confluentinc-kafka-connect-jdbc/lib
        curl -O https://jdbc.postgresql.org/download/postgresql-42.6.0.jar
        echo "Starting Kafka Connect..."
        /etc/confluent/docker/run
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:8083/"]
      interval: 15s
      timeout: 10s
      retries: 10

  # Snowpipe Streaming Client
  snowpipe-streaming:
    build:
      context: ./snowpipe
      dockerfile: Dockerfile
    container_name: snowpipe-streaming
    depends_on:
      - kafka
    env_file:
      - .env
    environment:
      KAFKA_BOOTSTRAP_SERVERS: kafka:9092
      KAFKA_GROUP_ID: snowpipe-consumer-group
    networks:
      - pos-network
    restart: unless-stopped

  # Connector Setup Service (run once to configure Kafka Connect)
  connector-setup:
    image: curlimages/curl:latest
    container_name: connector-setup
    depends_on:
      kafka-connect:
        condition: service_healthy
    volumes:
      - ./kafka-connect:/kafka-connect
    networks:
      - pos-network
    command:
      - /bin/sh
      - -c
      - |
        echo "Waiting for Kafka Connect to be fully ready..."
        sleep 30

        echo "Creating transaction CDC Source Connector..."
        curl -X POST http://kafka-connect:8083/connectors \
          -H "Content-Type: application/json" \
          -d @/kafka-connect/pos-source.json

        echo "Connector setup complete!"

        echo "Listing all connectors..."
        curl -X GET http://kafka-connect:8083/connectors

networks:
  pos-network:
    driver: bridge

volumes:
  pos_data:

Snowflake Setup

We need to create the proper database, schema and tables so store the ingestion result. This is a sample query we can use:

-- Create database
CREATE DATABASE IF NOT EXISTS DB_DEMO;
USE DATABASE DB_DEMO;

-- Create schemas
CREATE SCHEMA IF NOT EXISTS POS_DATA;

-- ============================================================================
-- POS_DATA Schema
-- ============================================================================
USE SCHEMA POS_DATA;

CREATE TABLE IF NOT EXISTS transactions (
    transaction_id INTEGER PRIMARY KEY,
    transaction_code VARCHAR(100),
    outlet_id INTEGER,
    transaction_date TIMESTAMP_NTZ,
    total_amount NUMBER(15, 2),
    total_discount NUMBER(15, 2),
    payment_method VARCHAR(50),
    cashier_name VARCHAR(255),
    customer_id VARCHAR(100),
    status VARCHAR(20),
    created_at TIMESTAMP_NTZ,
    snowflake_ingested_at TIMESTAMP_LTZ DEFAULT CURRENT_TIMESTAMP()
);

ALTER TABLE transactions ADD CONSTRAINT unique_transaction_code UNIQUE (transaction_code);

We also need to store the credentials in an .env file to enable the secure connection

SNOWFLAKE_ACCOUNT=ORGNAME-ACCOUNTNAME
SNOWFLAKE_USER=YOUR_USER
SNOWFLAKE_PASSWORD=USE_PAT_TOKEN
SNOWFLAKE_DATABASE=DB_DEMO
SNOWFLAKE_SCHEMA=POS_DATA
SNOWFLAKE_WAREHOUSE=YOUR_COMPUTE
SNOWFLAKE_ROLE=YOUR_ROLE

# Kafka Configuration
KAFKA_BOOTSTRAP_SERVERS=kafka:9092
KAFKA_GROUP_ID=snowpipe-consumer-group

# PostgreSQL - POS Database
POS_DB_HOST=pos-db
POS_DB_PORT=5432
POS_DB_NAME=pos_db
POS_DB_USER=your_username
POS_DB_PASSWORD=your_password

Create Shell Script

Once everything is ready, we can set create another script to run the whole process.

#!/bin/bash

# CDC Demo Staged Startup Script
# Starts services in stages for better reliability with dynamic waiting

set -e

# Helper function: Wait for service with health check
wait_for_service() {
    local SERVICE_NAME=$1
    local CHECK_COMMAND=$2
    local MAX_WAIT=${3:-120}
    local WAIT_TIME=0

    echo "⏳ Waiting for ${SERVICE_NAME} to be ready..."

    while [ $WAIT_TIME -lt $MAX_WAIT ]; do
        if eval "$CHECK_COMMAND" > /dev/null 2>&1; then
            echo "✓ ${SERVICE_NAME} is ready (waited ${WAIT_TIME}s)"
            return 0
        fi
        sleep 5
        WAIT_TIME=$((WAIT_TIME + 5))
        if [ $((WAIT_TIME % 15)) -eq 0 ]; then
            echo "  Still waiting... (${WAIT_TIME}s/${MAX_WAIT}s)"
        fi
    done

    echo "❌ Timeout: ${SERVICE_NAME} didn't start within ${MAX_WAIT} seconds"
    return 1
}

echo "======================================================================"
echo "  Legacy DB CDC Demo - Staged Startup"
echo "======================================================================"
echo ""

# Check Docker
if ! command -v docker &> /dev/null; then
    echo "❌ Error: Docker is not installed"
    echo "   Please install Docker from https://www.docker.com/get-started"
    exit 1
fi

if ! docker info &> /dev/null; then
    echo "❌ Error: Docker daemon is not running"
    echo "   Please start Docker Desktop"
    exit 1
fi

echo "✓ Docker is running"

# Check Docker Compose
if ! command -v docker-compose &> /dev/null; then
    echo "❌ Error: Docker Compose is not installed"
    exit 1
fi

echo "✓ Docker Compose is available"

# Check .env file
if [ ! -f .env ]; then
    echo ""
    echo "⚠️  Warning: .env file not found"
    echo "   Creating .env from env.example..."

    if [ -f env.example ]; then
        cp env.example .env
        echo "📝 Created .env from env.example"
        echo "   Please edit .env and add your Snowflake credentials:"
        echo "   - SNOWFLAKE_ACCOUNT"
        echo "   - SNOWFLAKE_USER"
        echo "   - SNOWFLAKE_PASSWORD"
        echo ""
        read -p "Press Enter after updating .env file..."
    else
        echo "❌ Error: env.example not found"
        exit 1
    fi
fi

echo "✓ .env file exists"

# Check Snowflake credentials
if grep -q "your-account" .env 2>/dev/null; then
    echo "⚠️  Warning: Snowflake credentials may not be configured"
    read -p "Continue anyway? (y/N) " -n 1 -r
    echo
    if [[ ! $REPLY =~ ^[Yy]$ ]]; then
        exit 1
    fi
else
    echo "✓ Snowflake credentials are configured"
fi

# Check if containers are already running
if docker-compose ps 2>/dev/null | grep -q "Up"; then
    echo ""
    echo "⚠️  Some containers are already running"
    read -p "Stop and restart all containers? (y/N) " -n 1 -r
    echo
    if [[ $REPLY =~ ^[Yy]$ ]]; then
        echo "Stopping existing containers..."
        docker-compose down
    else
        echo "Exiting..."
        exit 0
    fi
fi

echo ""
echo "======================================================================"
echo "  STAGE 1: Starting Kafka Infrastructure"
echo "======================================================================"
echo ""

echo "Starting Zookeeper, Kafka, and Kafka Connect..."
docker-compose up -d zookeeper kafka kafka-connect

# Wait for Kafka to be ready
if ! wait_for_service "Kafka broker" \
    "docker exec kafka kafka-broker-api-versions --bootstrap-server localhost:9092" \
    120; then
    echo "   Check logs: docker-compose logs kafka"
    exit 1
fi

# Wait for Kafka Connect to be ready
if ! wait_for_service "Kafka Connect" \
    "curl -s http://localhost:8083/" \
    120; then
    echo "   Check logs: docker-compose logs kafka-connect"
    exit 1
fi

echo "Registering Kafka Connect connectors..."
docker-compose up -d connector-setup

# Wait for connectors to be registered (with automatic fallback)
MAX_WAIT=120
WAIT_TIME=0
echo "⏳ Waiting for connector registration..."

while [ $WAIT_TIME -lt $MAX_WAIT ]; do
    CONNECTORS=$(curl -s http://localhost:8083/connectors 2>/dev/null || echo "[]")
    if echo "$CONNECTORS" | grep -q "pos-cdc-source"; then
        echo "✓ Connectors registered successfully (waited ${WAIT_TIME}s)"
        break
    fi
    sleep 5
    WAIT_TIME=$((WAIT_TIME + 5))
    if [ $((WAIT_TIME % 15)) -eq 0 ]; then
        echo "  Still waiting... (${WAIT_TIME}s/${MAX_WAIT}s)"
    fi
done

# If auto-registration failed, try manual registration
if [ $WAIT_TIME -ge $MAX_WAIT ]; then
    echo "⚠️  Auto-registration timed out, attempting manual registration..."

    # Manual registration of POS connector
    curl -X POST http://localhost:8083/connectors \
      -H "Content-Type: application/json" \
      -d @kafka-connect/pos-source.json > /dev/null 2>&1 || true

    sleep 5

    # Verify manual registration
    CONNECTORS=$(curl -s http://localhost:8083/connectors 2>/dev/null || echo "[]")
    if echo "$CONNECTORS" | grep -q "pos-cdc-source"; then
        echo "✓ Connectors manually registered successfully"
    else
        echo "❌ Failed to register connectors automatically"
        echo "   You can register manually later with:"
        echo "   curl -X POST http://localhost:8083/connectors -H 'Content-Type: application/json' -d @kafka-connect/pos-source.json"
    fi
fi

# Show registered connectors
REGISTERED=$(curl -s http://localhost:8083/connectors 2>/dev/null || echo "[]")
echo "📋 Registered connectors: $REGISTERED"

echo ""
echo "======================================================================"
echo "  STAGE 2: Starting Data Pipeline"
echo "======================================================================"
echo ""

echo "Starting Snowpipe Streaming..."
docker-compose up -d snowpipe-streaming

# Give generators a moment to initialize
echo "⏳ Waiting for generators to initialize..."
sleep 5

# Check if generators are running
if docker-compose ps | grep -q "snowpipe-streaming.*Up"; then
    echo "✓ Snowpipe streaming client is running"
else
    echo "⚠️  Warning: Snowpipe streaming may not have started properly"
fi

echo ""
echo "======================================================================"
echo "  ✓ All Stages Complete!"
echo "======================================================================"
echo ""
echo "Services Status:"
docker-compose ps

echo ""
echo "======================================================================"
echo "Monitoring Commands:"
echo "======================================================================"
echo ""
echo "Pipeline Status:"
echo "  • Snowpipe Streaming:    docker-compose logs -f snowpipe-streaming"
echo "  • Kafka Connect:         docker-compose logs -f kafka-connect"
echo ""
echo "Health Checks:"
echo "  • View all services:     docker-compose ps"
echo "  • Check connectors:      curl http://localhost:8083/connectors"
echo "  • Connector status:      curl http://localhost:8083/connectors/pos-cdc-source/status | jq"
echo "  • Check Kafka topics:    docker exec kafka kafka-topics --list --bootstrap-server localhost:9092"
echo "  • View CDC logs:         docker exec pos-db psql -U your_user -d pos_db -c 'SELECT COUNT(*) FROM transaction_cdc_log;'"
echo ""
echo "Database Access:"
echo "  • POS Database:          docker exec -it pos-db psql -U your_user -d pos_db"
echo "  • Or via local:          psql -h localhost -p 5434 -U your_user -d pos_db"
echo "  • Password:              your_password"
echo ""
echo "Next Steps:"
echo "  1. Verify CDC is working:"
echo "     docker exec kafka kafka-console-consumer --bootstrap-server localhost:9092 --topic pos-transaction_cdc_log --max-messages 5"
echo "  2. Monitor data flow:"
echo "     docker-compose logs -f snowpipe-streaming"
echo "  3. Check Snowflake: Run sample queries"
echo ""
echo "To stop the demo:"
echo "  docker-compose down"
echo ""
echo "To clean everything and start fresh:"
echo "  docker-compose down -v"
echo ""
echo "======================================================================"

Ingestion Result

Once we run the start_demo.sh script, all of the changes in our monitored table will be reflected in the Snowflake tables within seconds. I tried to simulate with 5 TPS for one and the latency is less than 10 seconds between the inserted data in PostgreSQL until it appears in Snowflake!

Potential Improvements

This approach provides a solution to deliver near real-time ingestion using a legacy database. In the future, we can add more detailed solution to cover the following:

Scalability

While this workaround can handle 5 TPS easily from a single table, a thorough testing is required to monitor the database performance like CPU and I/O. Another component to configure is the Kafka Connect, especially the number of workers to support expected latency and throughput in the production environment.

Observability

Several metrics may need to be captured when deploying this in production, such as lag behind the source DB, Kafka consumer lag, and Snowpipe Streaming error rates. The log can be stored and consumed by observability tools like Prometheus/Grafana, or other managed services.

Dynamic Function

For production scale, the function creation or trigger generation may require another automation process. By querying table information_schema which contains metadata, the supporting functions and triggers can be created dynamically without manual script development.

Conclusion

This article showed how we can provide a workaround when facing legacy systems. The approach chosen here may not be the most ideal way to ingest the data, but when upgrading the PostgreSQL version is not possible due to many factors, this approach works quite well to simulate the real-world workload. Please note that the additional table will consume more storage in your database, so we may need another scheduler to truncate the table periodically.

Hope you find this article useful!

How to Evaluate Your Text-to-SQL Agent in Cortex Analyst Using TruLens

Reza Brianca — Tue, 06 Jan 2026 00:05:00 +0000

Overview

Ever since the LLM caught attention in the data domain space, many enterprises have tried to reap the benefits of it. However, given the fast pace of the breakthrough in the foundation model and its ecosystems, many find it difficult to integrate many of these tools to fulfill their enterprise standard. One of the most popular discussions is related to enabling talk to your data capabilities. Many attempts have been made but the results are still difficult to quantify. It has a different approach with traditional machine learning where the evaluation methods are already in their mature state. In this article, I will share one of the approaches in LLM evaluation, particularly for text-to-sql agents.

The rise of Text-to-SQL AI

Let’s face the reality. Many of the analysts in the enterprise are struggling to fulfill the role as SQL translator. That is, converting business questions into an SQL where the data can come from many different sources. In addition, let’s admit that doing those mundane tasks is not really good for your long-term career. On the other hand, enterprises also see this converting process as the major bottleneck. There will be back and forth meetings to clarify the definition and ensure they got the correct numbers.

Now they all hope that LLM can solve this problem. While the critical component has been identified (namely semantic view or semantic model), we are still not sure how to include the evaluation process in the development cycles. The semantic model serves as an enterprise-wide governance layer for the Text-to-SQL agent. More specifically, this layer provides value with:

Enriching Context: Informs the LLM of business-friendly terms (e.g., "Monthly Revenue") and how they map to complex, underlying schema objects.
Enforce Compliance: Ensures the generated SQL adheres to predefined rules, preventing the LLM from accessing unauthorized or irrelevant data, which is crucial for security and compliance.

Given the nature of semantic models, you can see that this is different from the usual machine learning model evaluation like accuracy, recall, precision, RMSE, etc. The evaluation process mostly is still done manually with the analyst checking the accuracy of the generated SQL.

The Challenge: Run Evaluation in Scalable Way

Doing the evaluation process manually is a tedious process. Not only do analysts need to check the generated queries, they also need to check whether the SQL can produce the valid results. This is mainly because the output of SQL is a number or a table where it has to be correct without any deviations. For example, if the question is about monthly revenue, the result should select the valid columns to base the aggregate on and use the correct formula on the expected columns. This is the usual result where BI reporting can visualize it using a bar chart or line chart.

You can imagine if there are changes the following area:

Semantic model: we add more tables, remove irrelevant columns or modify the table or column definition
Response instruction: we modify response instruction to refine the output response
Verified queries: we add more sample valid queries to enrich the agent Then the evaluation process must be done again to ensure the agent still works as expected.

Thanks to the rapid development in this evaluation area, now we can use TruLens as one of the options to do this automatically. The high level approach is quite similar to the traditional evaluation process where we need to prepare an evaluation dataset. The main difference is the ground truth now is not a label or value but the expected SQL as the output.

TruLens for Cortex Analyst

The TruLens library is quite versatile so we can use this to evaluate many LLM-based processes. Snowflake already created a documentation and hands-on tutorial for RAG-based evaluation here. I modified this tutorial a bit so the same approach can be applied to Cortex Analyst (Snowflake’s text-to-sql capabilities).

Why this is important for AI team

Using TruLens in Snowflake is a key component for operationalizing and monitoring their Text-to-SQL models in production. The evaluation is not a one-time test but a continuous observability loop.
The evaluation process ensures they can rely on the generated SQL for critical reports, eliminating the manual, tedious validation step and speeding up the time-to-insight.

Libraries

We need to have use several libraries to enable this

import requests
import json
import pandas as pd
import os
from snowflake.snowpark import Session

# TruLens components imports
from trulens.apps.custom import TruCustomApp as TruApp
from trulens.connectors.snowflake import SnowflakeConnector
from trulens.core.otel.instrument import instrument
from trulens.otel.semconv.trace import SpanAttributes
from trulens.core.run import Run, RunConfig

If there are any missing libraries, we can install this using pip command.

pip install snowflake-snowpark-python pandas pyarrow requests trulens-core trulens-connectors-snowflake trulens-providers-cortex

Create CortexAnalystApp with TruLens Instrument

Set the configuration to create a connection to Snowflake from your local environment. For security reasons, please don’t put the credentials in the code. You can use environment variables or Snowflake Secrets (Medium, Documentation) to store this.

# ⚠️ CONFIGURATION

# Snowflake connection parameters
connection_parameters = {
    "account": "ORGNAME-ACCOUNTNAME", # using new format
    "user": "YOUR_USERNAME",
    "password": "YOUR_PASSWORD",
    "warehouse": "YOUR_COMPUTE",
    "database": "YOUR_DB",
    "schema": "YOUR_SCHEMA",
    "role": "YOUR_ROLE"
}

# PAT Token (create in Snowsight → Profile → Personal Access Tokens)
PAT_TOKEN = "YOUR_PAT_TOKEN"

# Semantic model path (must be uploaded to a Snowflake stage)
SEMANTIC_MODEL_PATH = "@YOUR_DB.YOUR_SCHEMA.YOUR_STAGE/YOUR_SEMANTIC.yaml"

# Evaluation dataset and observability configuration
EVAL_DATASET_TABLE = "YOUR_DB.YOUR_SCHEMA.ANALYST_EVAL_DATASET"  # Table with 'question' and 'expected_sql' columns

# You need to create new DB and schema
OBSERVABILITY_DB = "OBSERVABILITY_DB"
OBSERVABILITY_SCHEMA = "OBSERVABILITY_SCHEMA"

session = Session.builder.configs(connection_parameters).create()
print(f"✅ Session created: {session.get_current_database()}.{session.get_current_schema()}")

Semantic model samples based on the TPCH dataset can be downloaded from here. I use the TPCH dataset for this exercise. The dataset is readily available as SNOWFLAKE_SAMPLE_DATA when you sign up for a Snowflake account.

To help with the scalability, we can create the python class for easier function calls later.

class CortexAnalystApp:
    """
    Cortex Analyst application with TruLens instrumentation.
    Uses REST API for SQL generation and includes observability.
    """

    def __init__(self, session, semantic_model_path, pat_token):
        """
        Initialize the Cortex Analyst app.

        Args:
            session: Snowpark session for SQL execution
            semantic_model_path: Path to semantic model in Snowflake stage
            pat_token: Personal Access Token for authentication
        """
        if not pat_token:
            raise ValueError("❌ PAT token is required!")

        self.session = session
        self.semantic_model_path = semantic_model_path
        self.pat_token = pat_token

        # Get proper Snowflake account URL
        self.host = self._get_snowflake_host()
        self.url = f"https://{self.host}/api/v2/cortex/analyst/message"

        # Load semantic model for context
        self.schema_context = self._load_semantic_model()

        print(f"✅ CortexAnalystApp initialized with TruLens instrumentation")
        print(f"   Host: {self.host}")
        print(f"   URL: {self.url}")
        print(f"   Semantic Model: {self.semantic_model_path}")

    def _get_snowflake_host(self):
        """Get the proper Snowflake account URL using new format."""
        try:
            result = self.session.sql("SELECT CURRENT_ORGANIZATION_NAME() || '-' || CURRENT_ACCOUNT_NAME()").collect()
            account = result[0][0].lower().replace('_', '-')
            host = f"{account}.snowflakecomputing.com"

            return host
        except Exception as e:
            raise ValueError(f"Could not determine Snowflake host: {str(e)}")

    def _load_semantic_model(self):
        """Load semantic model from stage."""
        try:
            stream = self.session.file.get_stream(self.semantic_model_path)
            return stream.read().decode('utf-8')
        except Exception as e:
            print(f"⚠️ Could not load semantic model: {e}")
            return ""

    @instrument(
        span_type=SpanAttributes.SpanType.RETRIEVAL,
        attributes={SpanAttributes.RETRIEVAL.RETRIEVED_CONTEXTS: "return"}
    )
    def retrieve_context(self):
        """Retrieve semantic model as context for evaluation."""
        return [self.schema_context]

    @instrument(span_type=SpanAttributes.SpanType.GENERATION)
    def generate_sql(self, question: str) -> str:
        """
        Generate SQL using Cortex Analyst REST API.

        Args:
            question: Natural language question

        Returns:
            Generated SQL string
        """
        payload = {
            "messages": [
                {
                    "role": "user",
                    "content": [{"type": "text", "text": question}]
                }
            ],
            "semantic_model_file": self.semantic_model_path
        }

        headers = {
            "Authorization": f"Bearer {self.pat_token}",
            "X-Snowflake-Authorization-Token-Type": "PROGRAMMATIC_ACCESS_TOKEN",
            "Content-Type": "application/json"
        }

        try:
            response = requests.post(self.url, json=payload, headers=headers, timeout=60)

            if response.status_code != 200:
                error_msg = f"HTTP {response.status_code}: {response.text}"
                print(f"❌ API Error: {error_msg}")
                return f"Error: {error_msg}"

            resp_json = response.json()

            # Extract SQL from response
            if "message" in resp_json and "content" in resp_json["message"]:
                for item in resp_json["message"]["content"]:
                    if item.get("type") == "sql":
                        return item.get("statement", "")

            print(f"⚠️ Unexpected response format: {resp_json}")
            return "No SQL generated"

        except Exception as e:
            error_msg = str(e)
            print(f"❌ Error calling Cortex Analyst: {error_msg}")
            return f"Error: {error_msg}"

    def execute_sql(self, sql: str) -> str:
        """
        Execute generated SQL for validation.

        Args:
            sql: SQL query to execute

        Returns:
            String representation of results
        """
        if not sql or "Error" in sql:
            return "No valid SQL"
        try:
            df = self.session.sql(sql).to_pandas()
            return df.head(10).to_string()
        except Exception as e:
            return f"Execution Failed: {str(e)}"

    def query(self, question: str) -> str:
        """
        Main entry point - retrieves context, generates SQL, validates.
        This is the method that TruLens will automatically trace.

        Args:
            question: Natural language question

        Returns:
            Generated SQL string
        """
        # Retrieve context (semantic model)
        context = self.retrieve_context()

        # Generate SQL
        sql = self.generate_sql(question)

        # Execute for validation (optional, helps with evaluation)
        result = self.execute_sql(sql)

        return sql

# Initialize the app
analyst_app = CortexAnalystApp(session, SEMANTIC_MODEL_PATH, PAT_TOKEN)

You can test sample questions to test this class.

# Test query
test_question = "What was the total revenue in 1995?" # change this according to your question
print(f"📝 Question: {test_question}")

result_sql = analyst_app.query(test_question)
print(f"\n✅ Generated SQL:\n{'-'*60}")
print(result_sql)

Once the SQL can be generated from the sample question, we can continue to register this class into TruLens.

Register the App to TruLens

We need to register the class so the evaluation process can be captured properly inside the TruLens library and eventually will appear in the Snowflake platform.

# Switch to observability schema
session.use_schema(f"{OBSERVABILITY_DB}.{OBSERVABILITY_SCHEMA}") # Choose where to store the evaluation result
print(f"✅ Using: {session.get_current_database()}.{session.get_current_schema()}")

# Create TruLens connector for Snowflake
tru_snowflake_connector = SnowflakeConnector(snowpark_session=session)

# Register the app with TruLens
tru_analyst = TruApp(
    analyst_app,
    app_name="cortex_analyst_sql_gen",
    app_version="v1_trulens_eval",
    connector=tru_snowflake_connector,
    main_method=analyst_app.query  # Specify the main entry point method
)

print("✅ App registered with TruLens for observability")

Configure Evaluation Run

We need to have a table with at least 2 (two) columns that contains the questions and expected SQL statements.

# Configure the evaluation run
run_config = RunConfig(
    run_name="analyst_eval_run_1",
    dataset_name=EVAL_DATASET_TABLE,
    description="Questions about the TPCH-SF1",
    label="tpch_analyst_eval",
    source_type="TABLE",
    dataset_spec={
        "input": "QUESTION",            # Business questions
        "ground_truth_output": "EXPECTED_SQL",  # Valid SQL
    },
)

# Add the run to the app
run = tru_analyst.add_run(run_config=run_config)
print(f"✅ Run configured: {run_config.run_name}")
print(f"   Dataset: {EVAL_DATASET_TABLE}")

The configuration above basically tells TruLens to get the ground truth so later it can run the evaluation against it. Here are some samples from the evaluation dataset.
Question-1: What is the total revenue generated from all orders?

Expected SQL:

WITH __order_lineitems AS (
  SELECT
    l_shipdate AS lineitem_ship_date,
    l_extendedprice * (1 - l_discount) AS net_revenue
  FROM
    snowflake_sample_data.tpch_sf1.lineitem
)
SELECT
  MIN(ol.lineitem_ship_date) AS start_date,
  MAX(ol.lineitem_ship_date) AS end_date,
  SUM(ol.net_revenue) AS total_revenue
FROM
  __order_lineitems AS ol

Question-2: What is the monthly revenue trend?

Expected SQL:

WITH __order_lineitems AS (
  SELECT
    l_shipdate AS lineitem_ship_date,
    l_extendedprice * (1 - l_discount) AS net_revenue
  FROM
    snowflake_sample_data.tpch_sf1.lineitem
)
SELECT
  DATE_TRUNC('MONTH', lineitem_ship_date) AS month,
  MIN(lineitem_ship_date) AS start_date,
  MAX(lineitem_ship_date) AS end_date,
  SUM(net_revenue) AS monthly_net_revenue
FROM
  __order_lineitems
GROUP BY
  DATE_TRUNC('MONTH', lineitem_ship_date)
ORDER BY
  month DESC NULLS LAST

Question-3: What is the average part retail price by manufacturer?

Expected SQL:

WITH __parts AS (
  SELECT
    p_mfgr AS part_manufacturer_name,
    p_retailprice AS part_retail_price
  FROM
    snowflake_sample_data.tpch_sf1.part
)
SELECT
  p.part_manufacturer_name,
  AVG(p.part_retail_price) AS avg_retail_price
FROM
  __parts AS p
GROUP BY
  p.part_manufacturer_name
ORDER BY
  avg_retail_price DESC NULLS LAST

Then we can start the evaluation

# Start the evaluation run
print("🚀 Starting evaluation run...")

run.start()

print("✅ Evaluation run completed!")

The command above will run the evaluation process in Snowflake using TruLens library. If you run into error due to existing run_name already exists, you can use new name for the run_name or use this command to remove the previous run.

run = tru_analyst.get_run(run_name="analyst_eval_run_1")
run.delete()

Compute Evaluation Metrics

Once the evaluation process is complete, we still need to run the evaluation metrics to quantify the text-to-sql output. We can use several metrics to evaluate the quality of the agent.

# Compute evaluation metrics
print("📊 Computing evaluation metrics...")
print("   This may take a few minutes depending on dataset size\n")

metrics_to_compute = [
    "answer_relevance",    # How relevant is the SQL to the question
    "groundedness",        # Is SQL grounded in the semantic model
    "coherence",           # Is the SQL logically coherent
    "correctness"          # Is SQL correct compared to ground truth
]

print(f"   Metrics: {', '.join(metrics_to_compute)}\n")

try:
    run.compute_metrics(metrics_to_compute)
    print("✅ Metrics computed successfully!")
    print("   Results are stored in your observability database")
except Exception as e:
    print(f"⚠️ Error computing metrics: {e}")
    print("   Some metrics may have completed")

Depending on the evaluation dataset, this process can run for a while. Once the computing process is complete, you can see the results in the AI & ML → Cortex AI → Evaluations and select your app_name.

If you click on the evaluation run_name, it will provide the details on each question’s metrics result as defined in the metrics_to_compute parameter.

How to Interpret the Results

Here are some guidelines on what does each metrics measure in the evaluation process:

Low Correctness Score: Likely an issue with the semantic model (missing definitions or incorrect mappings) or a flaw in the evaluation dataset's Expected SQL. This requires an update to the semantic model or ground truth.
Low Groundedness Score: Indicates the model generated SQL that is trying to query entities outside of the defined semantic model. This is a critical governance failure that requires immediate attention and model tuning. These metrics are continuously logged back into Snowflake, providing a quantifiable metric for Regression Testing to ensure future model updates do not degrade performance on historical questions. ### Sample Results Here is a sample result difference from the question: “How many parts are supplied by suppliers from different regions than where they're ordered, grouped by the region pair?” Generated SQL

...
WHERE
  sr.region_name <> cr.region_name
GROUP BY
...

Expected SQL

...
WHERE
  sr.region_name <> cr.region_name
  AND NOT sr.region_name IS NULL
  AND NOT cr.region_name IS NULL
GROUP BY
...

Both generated the same CTEs in the process, the main difference is in the WHERE clause where the ground truth had additional column filters. TruLens saw this and marked the correctness metrics with 0.67

Conclusions

In this article, we have explored that using TruLens library can be integrated in your text-to-sql evaluation process. Although this example mainly uses Cortex Analyst, you can also explore this with different API endpoints and modify the generate_sql function. You can run this evaluation as part of your development process and enrich your evaluation dataset by adding more valid business questions and queries.

I hope this will give an inspiration on how to efficiently evaluate your talk-to-sql AI tools and prevent another bottleneck that may come in reaping the benefits of using AI at enterprise level.

Huge thanks to Vinut and Sheena for their guidance and review.

Feature Stores: The Secret Sauce for Real-Time ML (and Sanity) in Production

Reza Brianca — Fri, 17 Oct 2025 12:30:00 +0000

Ever opened your ride-hailing app, seen a surge price, and wondered how they calculate that so fast? Or perhaps as a data professional, you've spent weeks building a brilliant ML model in a notebook, only to see it falter in production due to subtle data inconsistencies. You're not alone. The gap between ML development and robust, real-time deployment is a common, often painful, challenge.

The unsung hero bridging this gap is often the Feature Store.

Let's dive into why it's a game-changer, especially for real-time applications, and how it simplifies MLOps.

The ML Lifecycle: A Quick Refresher

At a high level, the machine learning workflow typically involves:

Data Ingestion with Exploration: Gathering raw data (e.g., GPS coordinates, user clicks, weather data).
Feature Engineering: Transforming raw data into predictive signals (features). Think distance_to_nearest_driver from raw GPS.
Model Training: Teaching a model to find patterns in these features using historical data.
Model Deployment & Inference: Putting the model to work, making predictions on new, unseen data, often in real-time.
Model Monitoring: Keeping an eye on model performance and data drift in production.

Take a look at this typical machine learning infrastructure from a16z article.

You can access the high resolution image here.

The moment of truth usually hits between steps 3 and 4. The features used for training must be identical to those used for live predictions. Any deviation, however small, leads to training-serving skew, a silent model killer.

The Pain Points: Why MLOps Can Be a Nightmare

Before we introduce the solution, let's acknowledge the chaos many teams face without a proper Feature Store:

For Data Scientists: You build a fantastic feature like user_loyalty_score or local_event_density. You spend days perfecting it. Then an ML Engineer has to re-implement your carefully crafted Python logic into Java or Go for the production system. Inevitably, subtle differences creep in, leading to unpredictable model performance. You also waste time rebuilding features that already exist elsewhere.
For ML Engineers: You're stuck building complex data pipelines to push batch-computed features into a low-latency online store, manually backfilling historical data, and debugging data consistency issues across disparate systems. It's plumbing, not MLOps innovation.
For the Business: Delayed ML projects, inconsistent model performance, and an inability to iterate quickly on new features translate directly into missed opportunities and frustrated users.

This often leads to "feature anarchy", meaning a collection of ad-hoc SQL queries, scattered scripts, and duplicated feature logic across different repos, models, and teams. There's no single source of truth, and no easy way to discover or reuse what's already been built.

Enter the Feature Store: Your Single Source of Truth for ML Features 🎯

A Feature Store is a specialized data system designed to manage the entire lifecycle of your machine learning features. Think of it as a central hub where features are defined, stored, versioned, and served consistently for both training and inference.

It typically offers two key interfaces:

Offline Store: For serving large volumes of historical feature data for model training and backfilling.
Online Store: For serving individual feature vectors with ultra-low latency for real-time predictions.

Real-World Example: Dynamic Pricing in Ride-Hailing

Let's use the dynamic pricing example. A ride-hailing service needs to predict demand and supply imbalances in real-time to adjust prices. Your ML model might rely on features like:

avg_driver_eta_in_zone_5min: The average estimated time of arrival for drivers in a specific geographic zone over the last 5 minutes.
num_open_requests_in_zone_1min: Number of unfulfilled ride requests in the zone in the last minute.
is_holiday_or_special_event: A categorical feature indicating if it's a holiday or a major local event.
past_7d_peak_demand_hour: The hour of peak demand in that zone over the last 7 days.

To train your dynamic pricing model, you'd feed it historical values of these features (and many more) along with past pricing outcomes. This requires querying large datasets of historical feature values from your Offline Store.

Now, when a user opens the app, the pricing model needs to make an instant prediction. It needs the current values of avg_driver_eta_in_zone_5min, num_open_requests_in_zone_1min, etc. It queries the Online Store for these up-to-the-second features.

The Feature Store guarantees that the logic to compute avg_driver_eta_in_zone_5min for training is exactly the same as the logic used to fetch it for a real-time price prediction. This consistency is paramount for accurate, reliable models.

Furthermore, a feature store acts as a central catalog for these features. Data scientists can easily discover avg_driver_eta_in_zone_5min, see its definition, lineage, and who owns it. This prevents reinvention and fosters feature reuse across different models (e.g., perhaps a driver allocation model also uses this feature).

The Challenge: Building a Traditional Feature Store

The concept is compelling, but historically, implementing a feature store has been a significant engineering undertaking. The biggest blockers:

Dual Infrastructure: You typically need a robust, scalable data warehouse (e.g., for batch ETL and the Offline Store) AND a separate, ultra-low-latency key-value store (e.g., Redis, Cassandra) for the Online Store.
Data Synchronization: Keeping these two separate systems synchronized, ensuring data freshness, consistency, and backfilling historical data correctly, is a non-trivial engineering feat.
Point-in-Time Correctness: For training, you need to retrieve features as they were at the exact moment the historical event occurred. This "time travel" for features is complex to manage across distributed systems.

How Snowflake Simplifies the Feature Store (with a Disclaimer)

Disclaimer: I work at Snowflake and am genuinely excited about how our platform addresses these common MLOps pain points.

The core problem of a feature store is managing two drastically different data access patterns:

Large-scale historical batch processing during model training
Lightning-fast single-row lookups for model inference in production

Instead of stitching together separate systems, what if you could use a single, unified platform?

Snowflake Feature Store UI

Snowflake Feature Store Lineage

You can find the full documentation, including the high resolution image here.

This is where the Snowflake Data Cloud offers a compelling approach:

Unified Data Plane: Snowflake's architecture is designed to handle both massive analytical workloads (perfect for your Offline Store) and increasingly fast, low-latency queries (less than 100ms). Features like Search Optimization Service and Hybrid Tables (Unistore) mean you can leverage a single copy of your data to serve both training and inference needs. No more complex ETL pipelines to synchronize two disparate databases.
Zero-Copy Clones with Time Travel: This is a game-changer for ML. Need features as they existed for a training run last Thursday at 10 AM? Snowflake's Time Travel allows you to query data as it was at any point in the past. Combine this with Zero-Copy Clones to instantly create isolated copies of your feature tables for experimentation without impacting production or incurring storage costs. This inherently solves point-in-time correctness.
Snowpark for Feature Engineering: You can write all your complex feature engineering logic in Python, Java, or Scala using Snowpark, and execute it directly within Snowflake. Your feature pipelines run where your data lives, eliminating costly data movement and simplifying your architecture. You can generate features once and use them for both online and offline purposes.
Built-in Governance & Sharing: Snowflake's robust security, access controls, and data sharing capabilities mean your feature store isn't just a collection of data; it's a governed asset. You can control who sees what, track lineage, and easily share features across teams or even external partners.

By leveraging these capabilities, you spend less time on complex data engineering and more time building, deploying, and iterating on more accurate, reliable ML models.

Ready to Elevate Your MLOps? 🚀

A Feature Store is no longer a luxury; it's a necessity for any organization serious about scaling machine learning in production, especially for real-time use cases like dynamic pricing. Modern data platforms like Snowflake are dramatically lowering the barrier to entry, making it feasible for more teams to implement this critical piece of MLOps infrastructure.

Want to get hands-on and see how this works? Check out the Guide to Snowflake Feature Store Quickstart. It walks you through building a complete ML workflow from feature engineering until model deployment, all within a unified environment.

What are your biggest MLOps pain points? Share your thoughts or questions in the comments!