Designing and Implementing a Real-Time Ingestion Pipeline with Kafka, dbt, and Snowflake Dynamic Tables for Sub-Minute Analytics Freshness
Introduction
Your fraud detection dashboard refreshes every 15 minutes. A fraudulent transaction hits your platform at 2:03 PM. Your analyst sees it at 2:18 PM — and by then, the account has already been drained, a chargeback has been filed, and the damage is done. This is the real cost of batch latency, and it's a problem that exists across industries: e-commerce, fintech, logistics, healthcare ops. The gap between when data exists and when your organization can act on it is often measured in minutes or hours, even on stacks that are otherwise considered "modern."
The traditional fix was to throw Apache Spark Streaming or a dedicated OLAP engine like Druid at the problem and accept a parallel, specialized infrastructure. That works, but it creates bifurcated lineage, duplicate transformation logic, and a governance nightmare. The newer approach — the one we're going to build together in this lesson — threads real-time ingestion through Kafka directly into Snowflake, then uses Snowflake Dynamic Tables and carefully designed dbt models to deliver sub-minute analytics freshness without abandoning the transformation patterns your team already knows.
By the end of this lesson, you'll have a production-grade reference architecture and working code for a pipeline that ingests event streams from Kafka into Snowflake's raw layer, applies medallion-style transformations using Dynamic Tables, and surfaces clean, queryable marts with freshness guarantees under 60 seconds. You'll also understand where the sharp edges are — because there are plenty of them.
What you'll learn:
- How to design a Kafka → Snowflake ingestion path using the Snowflake Kafka Connector and Snowpipe Streaming
- How Snowflake Dynamic Tables work internally, their refresh mechanics, and how to configure target lag for sub-minute freshness
- How to model the transformation layer using dbt alongside Dynamic Tables without creating conflicting materialization strategies
- How to handle late-arriving data, schema evolution, and backpressure at each layer
- How to monitor pipeline health and set data freshness SLAs that actually mean something
Prerequisites
This lesson assumes you're comfortable with the following:
Why This Architecture, and Why Now
Before writing a single line of code, it's worth understanding the design space. Sub-minute analytics freshness can be achieved many ways, and choosing the right one for your context matters enormously.
If you've been wrestling with the tradeoffs between streaming and batch, the Real-Time Data: When to Use Streaming vs Batch Processing lesson covers this decision framework in depth. The short version for our purposes: when your downstream consumers need to act on data within seconds or low minutes — fraud alerts, live dashboards, real-time personalization — streaming ingestion is warranted. When they need to analyze trends over historical windows, batch is usually cheaper and simpler. This lesson targets the former, but we'll design the pipeline so it doesn't break the latter.
The architecture we're building has three conceptually distinct layers:
Ingestion Layer (Kafka → Snowflake Bronze): Raw events land in Snowflake with minimal latency using Snowpipe Streaming, which operates on a continuous micro-batch model rather than scheduled file loads. This is fundamentally different from traditional Snowpipe, and understanding that distinction will save you hours of debugging.
Transformation Layer (Dynamic Tables, Bronze → Silver → Gold): Snowflake Dynamic Tables handle the incremental refresh logic automatically. Unlike scheduled tasks or Streams + Tasks patterns, Dynamic Tables declare what you want the output to look like and let Snowflake figure out how to keep it fresh. dbt sits alongside this layer to manage the transformation DAG, documentation, and testing.
Serving Layer (Gold tables + Semantic Layer): Clean, queryable marts that BI tools and APIs can hit directly, with freshness guarantees you can monitor and alert on.
Key insight: The reason this architecture is compelling isn't that it's the only way to achieve sub-minute freshness — it's that it achieves sub-minute freshness within your existing Snowflake investment, using transformation patterns your analytics engineers already know. You're not bolting on a separate streaming system; you're extending one system to do more.
The Ingestion Layer: Kafka to Snowflake Bronze
Snowpipe Streaming vs Traditional Snowpipe
Most Snowflake practitioners are familiar with Snowpipe, which triggers file loads from cloud storage stages (S3, GCS, Azure Blob). The latency floor for traditional Snowpipe is roughly 1-5 minutes — you're waiting for files to accumulate, a notification to fire, and a load queue to process.
Snowpipe Streaming is a completely different animal. It uses the insertRows() API (exposed through the Kafka connector and the Snowflake Ingest SDK) to write rows directly into Snowflake's server-side buffer, bypassing the file staging step entirely. Data becomes queryable in the target table typically within 1-10 seconds of the insertRows() call completing. This is what makes sub-minute freshness achievable at the ingestion layer.
The tradeoff is that Snowpipe Streaming generates more, smaller micro-partitions, which can create file compaction overhead over time. Snowflake runs automatic compaction, but on high-volume topics you should plan for this and size your tables' clustering accordingly.
Setting Up the Snowflake Kafka Connector
The Snowflake Kafka Connector (version 2.1+) supports Snowpipe Streaming as a delivery method. Here's a production-ready connector configuration for a payment events topic:
{
"name": "snowflake-payments-sink",
"config": {
"connector.class": "com.snowflake.kafka.connector.SnowflakeSinkConnector",
"tasks.max": "4",
"topics": "payments.events.v1",
"snowflake.url.name": "yourorg-youraccountid.snowflakecomputing.com:443",
"snowflake.user.name": "KAFKA_SERVICE_USER",
"snowflake.private.key": "${file:/opt/kafka/secrets/snowflake.properties:private.key}",
"snowflake.private.key.passphrase": "${file:/opt/kafka/secrets/snowflake.properties:private.key.passphrase}",
"snowflake.database.name": "RAW_DB",
"snowflake.schema.name": "KAFKA_INGEST",
"snowflake.role.name": "KAFKA_LOADER_ROLE",
"snowflake.ingestion.method": "SNOWPIPE_STREAMING",
"snowflake.enable.schematization": "true",
"key.converter": "org.apache.kafka.connect.storage.StringConverter",
"value.converter": "com.snowflake.kafka.connector.records.SnowflakeAvroConverter",
"value.converter.schema.registry.url": "http://schema-registry:8081",
"buffer.count.records": "10000",
"buffer.flush.time": "10",
"buffer.size.bytes": "5242880",
"errors.tolerance": "all",
"errors.deadletterqueue.topic.name": "payments.events.v1.dlq",
"errors.deadletterqueue.topic.replication.factor": "3",
"errors.log.enable": "true"
}
}
A few things in this configuration deserve explanation:
snowflake.ingestion.method: SNOWPIPE_STREAMING — This is the critical switch. Without it, the connector defaults to the file-based Snowpipe approach.
snowflake.enable.schematization: true — This tells the connector to flatten the JSON payload into individual columns rather than dumping the entire record into a RECORD_CONTENT variant column. For sub-minute freshness, you want schematized tables because columnar scans on typed data are dramatically faster than parsing VARIANT on every query.
buffer.flush.time: 10 — This sets the maximum time (in seconds) before buffered records are flushed to Snowflake. Combined with buffer.count.records and buffer.size.bytes, whichever threshold is hit first triggers a flush. 10 seconds is a reasonable starting point — lower values increase Snowflake API calls, higher values increase latency.
Dead letter queue configuration — Never skip this in production. Schema mismatches, serialization errors, and transient network issues will all cause individual records to fail. Without a DLQ, those records are silently dropped or cause the connector to halt.
Warning: Snowpipe Streaming uses Snowflake credits differently than query credits. Streaming ingestion consumes "Snowpipe Streaming" credits, which are separate from your compute warehouse credits. Before going to production, verify your Snowflake contract includes this feature and understand its pricing model — it's billed per row rather than per warehouse-second.
Preparing the Snowflake Landing Zone
Before starting the connector, set up your Snowflake objects with the right structure:
-- Create a dedicated database and schema for raw ingestion
CREATE DATABASE IF NOT EXISTS RAW_DB;
CREATE SCHEMA IF NOT EXISTS RAW_DB.KAFKA_INGEST;
-- Create the service account and role
CREATE ROLE KAFKA_LOADER_ROLE;
CREATE USER KAFKA_SERVICE_USER
DEFAULT_ROLE = KAFKA_LOADER_ROLE
DEFAULT_WAREHOUSE = KAFKA_WH
RSA_PUBLIC_KEY = 'MII...your_public_key...'; -- key-pair auth, never passwords
-- Grant minimum necessary privileges
GRANT USAGE ON DATABASE RAW_DB TO ROLE KAFKA_LOADER_ROLE;
GRANT USAGE ON SCHEMA RAW_DB.KAFKA_INGEST TO ROLE KAFKA_LOADER_ROLE;
GRANT CREATE TABLE ON SCHEMA RAW_DB.KAFKA_INGEST TO ROLE KAFKA_LOADER_ROLE;
-- The connector will auto-create tables, but if you want to pre-define them:
CREATE TABLE IF NOT EXISTS RAW_DB.KAFKA_INGEST.PAYMENTS_EVENTS_V1 (
PAYMENT_ID VARCHAR,
CUSTOMER_ID VARCHAR,
MERCHANT_ID VARCHAR,
AMOUNT_CENTS NUMBER,
CURRENCY_CODE VARCHAR(3),
PAYMENT_METHOD VARCHAR,
STATUS VARCHAR,
METADATA VARIANT, -- flexible bag for fields that vary
EVENT_TIMESTAMP TIMESTAMP_TZ,
KAFKA_OFFSET NUMBER,
KAFKA_PARTITION NUMBER,
KAFKA_TOPIC VARCHAR,
RECORD_METADATA VARIANT, -- Snowflake connector metadata
INGESTED_AT TIMESTAMP_NTZ DEFAULT CURRENT_TIMESTAMP()
)
CLUSTER BY (DATE_TRUNC('hour', EVENT_TIMESTAMP));
The clustering key on DATE_TRUNC('hour', EVENT_TIMESTAMP) is intentional. Most queries against this table will filter by time range, and micro-partition pruning on a time-based cluster dramatically reduces the amount of data scanned. Without clustering, Snowflake has to scan all the small files created by Snowpipe Streaming — which can get expensive fast.
Tip: Pre-create your tables with explicit column definitions rather than letting the connector auto-create them. Auto-created tables use VARCHAR(16777216) for all string columns and don't get clustering keys. It's much harder to retrofit these after the fact because Dynamic Tables that depend on the table won't automatically pick up structural changes.
Handling Schema Evolution
Payment events evolve. New fields get added, types change, occasionally fields get renamed. With Avro and the Schema Registry, you have backward and forward compatibility guarantees at the serialization layer — but you still need to handle the Snowflake side.
The Snowflake Kafka Connector with schematization handles additive schema changes automatically: new fields in the Avro schema result in new columns being added to the Snowflake table via ALTER TABLE ... ADD COLUMN. This happens without connector restarts and without data loss.
Non-additive changes — field removals, type changes — require more care. A type change from INT to LONG in Avro is backward compatible at the Avro level but may not automatically resolve to the correct Snowflake column type. The safest approach is to version your topics (payments.events.v1, payments.events.v2) and run parallel connectors during migrations, then cut downstream Dynamic Tables over to the new version.
For deeper treatment of schema contracts between producers and consumers, see Implementing Data Contracts Between Ingestion and Transformation: Defining, Enforcing, and Evolving Schemas Across Producer and Consumer Teams — the tooling covered there (Confluent Schema Registry with compatibility modes, contract testing) applies directly to this architecture.
This is where the architecture gets interesting. Dynamic Tables are Snowflake's answer to the "I want incremental, event-driven transformation without writing pipeline code" problem. Understanding how they work internally is essential for using them effectively.
What Dynamic Tables Actually Do
A Dynamic Table is defined by a SQL query, a target lag, and a warehouse. Snowflake maintains the table's contents by periodically recomputing rows that have changed since the last refresh. When you create a Dynamic Table, Snowflake:
- Executes the defining query once to populate the initial state
- Creates internal change-tracking metadata on the upstream tables
- Schedules automatic refreshes that use the upstream change data (via Snowflake Streams under the hood) to apply only the delta — not a full recompute
The target lag is the key parameter. Setting TARGET_LAG = '1 minute' tells Snowflake to try to keep the Dynamic Table no more than 1 minute behind its source tables. Note "try" — this is a target, not a guarantee. If your upstream table is receiving data faster than the warehouse can process incremental refreshes, lag will exceed the target.
Key insight: Dynamic Tables use Snowflake Streams internally to capture changes on their source tables. This means they inherit the same limitation: Streams have a data retention window (7 days by default), and if a Dynamic Table fails to refresh within that window due to an extended outage, the Stream may expire and require a full recompute on recovery. Monitor your Dynamic Table refresh history to catch this before it becomes a crisis.
Building the Bronze → Silver Dynamic Table
Let's build the Silver layer transformation for our payments pipeline. The Silver layer typically handles deduplication, type casting, and light business logic normalization — it's the "cleaned" version of the raw data.
CREATE OR REPLACE DYNAMIC TABLE ANALYTICS_DB.SILVER.PAYMENTS_SILVER
TARGET_LAG = '30 seconds'
WAREHOUSE = TRANSFORM_WH_XS
AS
SELECT
-- Core identifiers
PAYMENT_ID,
CUSTOMER_ID,
MERCHANT_ID,
-- Clean and normalize amounts
AMOUNT_CENTS / 100.0 AS amount_usd,
UPPER(TRIM(CURRENCY_CODE)) AS currency_code,
-- Normalize payment method categories
CASE
WHEN PAYMENT_METHOD ILIKE '%visa%' THEN 'VISA'
WHEN PAYMENT_METHOD ILIKE '%mastercard%' THEN 'MASTERCARD'
WHEN PAYMENT_METHOD ILIKE '%amex%' THEN 'AMEX'
WHEN PAYMENT_METHOD ILIKE '%paypal%' THEN 'PAYPAL'
ELSE 'OTHER'
END AS payment_method_category,
-- Status normalization and validation
UPPER(STATUS) AS status,
STATUS IN ('COMPLETED', 'FAILED',
'PENDING', 'REFUNDED') AS is_known_status,
-- Time fields
EVENT_TIMESTAMP AS event_at,
DATE_TRUNC('hour', EVENT_TIMESTAMP) AS event_hour,
DATE_TRUNC('day', EVENT_TIMESTAMP) AS event_date,
-- Kafka provenance (keep for debugging)
KAFKA_OFFSET,
KAFKA_PARTITION,
INGESTED_AT,
CURRENT_TIMESTAMP() AS silver_transformed_at
FROM RAW_DB.KAFKA_INGEST.PAYMENTS_EVENTS_V1
-- Deduplication: keep only the latest record per payment_id
QUALIFY ROW_NUMBER() OVER (
PARTITION BY PAYMENT_ID
ORDER BY EVENT_TIMESTAMP DESC, KAFKA_OFFSET DESC
) = 1;
A few architectural decisions embedded in this SQL:
Deduplication with QUALIFY: Kafka delivery semantics are at-least-once by default. Duplicate events will arrive, especially during connector restarts or rebalances. The QUALIFY ROW_NUMBER() pattern handles this cleanly in a single scan. We use KAFKA_OFFSET as a tiebreaker for events with identical timestamps.
Keeping provenance columns: KAFKA_OFFSET, KAFKA_PARTITION, and INGESTED_AT are preserved in Silver. When something goes wrong — and it will — you need a way to trace a specific bad record back to its exact position in the Kafka topic so you can replay or skip it.
TARGET_LAG of 30 seconds for Silver: We set this tighter than Gold because each downstream Dynamic Table adds its own lag budget. If Silver takes 30 seconds and Gold takes 30 seconds, our end-to-end lag is ~60 seconds. Design your lag targets as a budget distributed across layers.
Building the Gold Layer
The Gold layer materializes business-level aggregates that BI tools query directly:
CREATE OR REPLACE DYNAMIC TABLE ANALYTICS_DB.GOLD.PAYMENTS_METRICS_REALTIME
TARGET_LAG = '30 seconds'
WAREHOUSE = TRANSFORM_WH_XS
AS
SELECT
merchant_id,
payment_method_category,
currency_code,
event_hour,
-- Volume metrics
COUNT(*) AS total_transactions,
COUNT(*) FILTER (WHERE status = 'COMPLETED') AS completed_transactions,
COUNT(*) FILTER (WHERE status = 'FAILED') AS failed_transactions,
-- Amount metrics
SUM(amount_usd) FILTER (WHERE status = 'COMPLETED') AS gmv_usd,
AVG(amount_usd) FILTER (WHERE status = 'COMPLETED') AS avg_transaction_usd,
PERCENTILE_CONT(0.95) WITHIN GROUP (
ORDER BY amount_usd
) FILTER (WHERE status = 'COMPLETED') AS p95_transaction_usd,
-- Derived metrics
ROUND(
100.0 * COUNT(*) FILTER (WHERE status = 'FAILED')
/ NULLIF(COUNT(*), 0),
2) AS failure_rate_pct,
-- Freshness tracking
MAX(event_at) AS latest_event_at,
CURRENT_TIMESTAMP() AS gold_refreshed_at
FROM ANALYTICS_DB.SILVER.PAYMENTS_SILVER
GROUP BY 1, 2, 3, 4;
Warning: Dynamic Tables with aggregations have a critical limitation: they perform full recomputes of the aggregate groups that contain changed rows, not row-level incremental updates. For a COUNT(*) grouped by merchant_id and event_hour, any new payment event in hour 2024-01-15 14:00:00 causes a full recompute of that group. This is usually fine for sub-minute latency, but if a single group contains millions of rows and you're on a small warehouse, your refresh time will blow past your target lag. Size your warehouse accordingly and monitor refresh duration.
Chaining Dynamic Tables: The Dependency Graph
When you chain Dynamic Tables (Silver reads from Bronze table, Gold reads from Silver), Snowflake is smart enough to coordinate their refresh schedules. If Silver hasn't refreshed yet, Gold won't execute a refresh that would produce stale results. Snowflake tracks the dependency graph and propagates refreshes in topological order.
You can inspect this with:
-- Check refresh history and actual lag for all Dynamic Tables
SELECT
name,
target_lag,
scheduling_state,
last_completed_refresh,
DATEDIFF('second', last_completed_refresh, CURRENT_TIMESTAMP()) AS seconds_since_refresh,
last_completed_refresh_status
FROM INFORMATION_SCHEMA.DYNAMIC_TABLES
WHERE schema_name = 'SILVER' OR schema_name = 'GOLD'
ORDER BY name;
-- Detailed refresh performance
SELECT
name,
refresh_start_time,
refresh_end_time,
DATEDIFF('millisecond', refresh_start_time, refresh_end_time) AS duration_ms,
rows_inserted,
rows_deleted,
bytes_inserted,
state,
state_message
FROM INFORMATION_SCHEMA.DYNAMIC_TABLE_REFRESH_HISTORY(
NAME => 'ANALYTICS_DB.GOLD.PAYMENTS_METRICS_REALTIME'
)
ORDER BY refresh_start_time DESC
LIMIT 50;
Watch duration_ms closely in the first days after launch. If your refresh duration consistently approaches or exceeds your target lag interval, you're at risk of falling behind. Common causes: warehouse too small, GROUP BY on high-cardinality columns without clustering, or downstream Dynamic Tables blocking upstream refreshes due to concurrency limits.
Where dbt Fits In This Architecture
Here's the honest architectural tension: Snowflake Dynamic Tables and dbt both want to manage your materialization strategy, and they don't fully speak the same language yet. Let's be precise about where each belongs.
The Hybrid Strategy
The approach that works best in production is a hybrid model where:
- Dynamic Tables own the streaming transformation path (Bronze raw → Silver clean → Gold real-time aggregates)
- dbt owns documentation, testing, and any batch-style marts that sit downstream of the Gold Dynamic Tables
dbt has a dynamic_table materialization as of dbt-snowflake 1.6+. You can define a Dynamic Table in dbt like this:
# dbt_project.yml
models:
your_project:
streaming:
silver:
+materialized: dynamic_table
+snowflake_dynamic_table:
target_lag: "30 seconds"
snowflake_warehouse: TRANSFORM_WH_XS
And in your model SQL (models/streaming/silver/payments_silver.sql):
{{
config(
materialized='dynamic_table',
snowflake_dynamic_table={
'target_lag': '30 seconds',
'snowflake_warehouse': 'TRANSFORM_WH_XS'
}
)
}}
SELECT
PAYMENT_ID,
CUSTOMER_ID,
MERCHANT_ID,
AMOUNT_CENTS / 100.0 AS amount_usd,
-- ... rest of transformation
CURRENT_TIMESTAMP() AS silver_transformed_at
FROM {{ source('kafka_ingest', 'payments_events_v1') }}
QUALIFY ROW_NUMBER() OVER (
PARTITION BY PAYMENT_ID
ORDER BY EVENT_TIMESTAMP DESC
) = 1
When you run dbt run, dbt will execute CREATE OR REPLACE DYNAMIC TABLE ... with the correct TARGET_LAG and WAREHOUSE configuration. This gives you all the benefits of dbt's ecosystem — dbt test, dbt docs generate, lineage graphs — while using Dynamic Tables as the materialization engine.
Tip: When using dbt's dynamic_table materialization, dbt run will create or replace the Dynamic Table, which forces a full recompute on the first run after any code change. In production, use dbt run --full-refresh only when you need it, and be aware that replacing a Dynamic Table breaks any external Streams that were reading its change data. Coordinate with downstream consumers before running full refreshes on shared tables.
dbt Tests on Streaming Data
Testing streaming data in dbt requires a slightly different mindset than testing batch models. The data is always changing, which means some tests that would be deterministic in batch mode are inherently non-deterministic on live streaming tables.
# models/streaming/silver/schema.yml
version: 2
models:
- name: payments_silver
description: "Cleaned and deduplicated payment events from Kafka ingestion"
config:
materialized: dynamic_table
columns:
- name: payment_id
description: "Unique payment identifier"
tests:
- unique
- not_null
- name: amount_usd
tests:
- not_null
- dbt_utils.accepted_range:
min_value: 0
max_value: 100000 # sanity check - flag payments over $100k
- name: status
tests:
- accepted_values:
values: ['COMPLETED', 'FAILED', 'PENDING', 'REFUNDED']
quote: true
- name: currency_code
tests:
- not_null
sources:
- name: kafka_ingest
database: RAW_DB
schema: KAFKA_INGEST
freshness:
warn_after: {count: 2, period: minute}
error_after: {count: 5, period: minute}
loaded_at_field: INGESTED_AT
tables:
- name: payments_events_v1
The freshness configuration on the source is critical. This makes dbt source freshness check that data has arrived within the last 2 minutes (warn) or 5 minutes (error). For a deeper look at implementing freshness SLAs and alerting, Automating Data Freshness SLAs: Defining, Measuring, and Alerting on Staleness Across Your Modern Data Stack covers this comprehensively.
Late-Arriving Data and Out-of-Order Events
Kafka doesn't guarantee global ordering across partitions. A payment event with EVENT_TIMESTAMP = 2024-01-15 14:03:22 might arrive after an event with EVENT_TIMESTAMP = 2024-01-15 14:03:45 if they were on different partitions. More dramatically, mobile applications often buffer events locally and flush them in batches, which can mean events arriving minutes or hours after their actual timestamp.
Your Dynamic Table aggregations need to account for this. The naive approach — WHERE event_date = CURRENT_DATE() — will silently exclude late arrivals for previous days.
For the Gold aggregation table, consider a watermark-based windowing approach:
CREATE OR REPLACE DYNAMIC TABLE ANALYTICS_DB.GOLD.PAYMENTS_HOURLY_WITH_LATE_ARRIVALS
TARGET_LAG = '1 minute'
WAREHOUSE = TRANSFORM_WH_XS
AS
SELECT
merchant_id,
event_hour,
payment_method_category,
COUNT(*) AS total_transactions,
SUM(amount_usd) FILTER (WHERE status = 'COMPLETED') AS gmv_usd,
MAX(ingested_at) AS latest_ingestion_at,
MAX(event_at) AS latest_event_at,
CURRENT_TIMESTAMP() AS last_refreshed_at,
-- Track how many late arrivals we're seeing
COUNT(*) FILTER (
WHERE ingested_at > event_at + INTERVAL '5 minutes'
) AS late_arrival_count
FROM ANALYTICS_DB.SILVER.PAYMENTS_SILVER
-- Only aggregate hours that are "settled" or in the active window
-- Active window: last 2 hours (to capture late arrivals)
-- Historical data is handled by your batch dbt models
WHERE event_hour >= DATE_TRUNC('hour', CURRENT_TIMESTAMP() - INTERVAL '2 hours')
GROUP BY 1, 2, 3;
The WHERE event_hour >= ... clause creates a sliding window of the last 2 hours. Every Dynamic Table refresh rescans and recomputes all hour-level groups within that window, automatically incorporating late arrivals. Hours that fall outside the 2-hour window are frozen in place — their aggregates won't update for late arrivals.
Note: This sliding window approach means your Dynamic Table is doing more work per refresh than a pure append-only model. You're recomputing 2 hours × however many distinct (merchant_id, payment_method_category) combinations exist on every refresh. For very high cardinality dimensions, consider narrowing the window or pre-aggregating in Silver before reaching Gold.
For truly late data (more than 2 hours), you'll want a complementary batch dbt incremental model that runs hourly or daily to catch up:
-- models/batch/gold/payments_hourly_complete.sql
{{
config(
materialized='incremental',
unique_key=['merchant_id', 'event_hour', 'payment_method_category'],
incremental_strategy='merge',
cluster_by=['event_hour']
)
}}
SELECT
merchant_id,
event_hour,
payment_method_category,
COUNT(*) AS total_transactions,
SUM(amount_usd) FILTER (WHERE status = 'COMPLETED') AS gmv_usd
FROM {{ ref('payments_silver') }}
{% if is_incremental() %}
-- Reprocess last 6 hours to capture late arrivals and corrections
WHERE event_hour >= DATE_TRUNC('hour', CURRENT_TIMESTAMP() - INTERVAL '6 hours')
{% endif %}
GROUP BY 1, 2, 3
This creates a two-tier freshness model: the Dynamic Table provides sub-minute freshness for the recent window, and the batch dbt incremental model provides corrected, complete data for the historical record. Dashboard queries can choose which table to hit based on what they need. For incremental model design patterns, including handling late arrivals systematically, Incremental Models at Scale: Strategies for Efficiently Processing Late-Arriving Data and Partition Pruning in dbt is the companion piece to this lesson.
Warehouse Sizing for Dynamic Table Refreshes
Dynamic Tables run on dedicated virtual warehouses. For the sub-minute refresh targets we're using, the warehouse must start and execute the refresh query in under 30 seconds (ideally much less). This has two implications:
- Use X-Small or Small warehouses for simple filter-and-cast Silver transformations. The overhead of spinning up a larger warehouse often exceeds the benefit.
- Keep your Dynamic Table warehouses separate from your query warehouses. If analysts are hammering a Large warehouse with complex queries, and your Dynamic Table refresh shares that warehouse, you'll see refresh queuing. Use a dedicated
TRANSFORM_WH_XS just for Dynamic Table refreshes.
Warning: Snowflake auto-suspend is dangerous for Dynamic Tables on small warehouses with 30-second target lag. If the warehouse auto-suspends and has a 60-second resume time, every resume cycle costs you a full minute of lag. Set AUTO_SUSPEND = 60 (the minimum) and consider AUTO_RESUME = TRUE carefully — in practice, for sub-minute targets, you may need to disable auto-suspend entirely and accept the cost of a continuously running XS warehouse (~$1.20/hour on most Snowflake contracts). For cost management strategies in this context, see Cost Management in Cloud Data Platforms.
Monitoring with Snowflake's Dynamic Table Observability
Build a monitoring query you can schedule as a Snowflake Task or feed into your alerting system:
-- Pipeline health dashboard query
WITH refresh_stats AS (
SELECT
name,
DATEDIFF('second', last_completed_refresh, CURRENT_TIMESTAMP()) AS actual_lag_seconds,
CASE target_lag
WHEN '30 seconds' THEN 30
WHEN '1 minute' THEN 60
WHEN '5 minutes' THEN 300
ELSE NULL
END AS target_lag_seconds,
scheduling_state,
last_completed_refresh_status
FROM INFORMATION_SCHEMA.DYNAMIC_TABLES
WHERE database_name = 'ANALYTICS_DB'
),
refresh_durations AS (
SELECT
name,
AVG(DATEDIFF('millisecond', refresh_start_time, refresh_end_time)) AS avg_refresh_ms,
MAX(DATEDIFF('millisecond', refresh_start_time, refresh_end_time)) AS max_refresh_ms,
COUNT(*) FILTER (WHERE state = 'FAILED') AS failed_refreshes_last_50
FROM INFORMATION_SCHEMA.DYNAMIC_TABLE_REFRESH_HISTORY(
ERROR_ONLY => FALSE
)
GROUP BY 1
)
SELECT
r.name,
r.actual_lag_seconds,
r.target_lag_seconds,
r.actual_lag_seconds - r.target_lag_seconds AS lag_overage_seconds,
d.avg_refresh_ms / 1000.0 AS avg_refresh_sec,
d.max_refresh_ms / 1000.0 AS max_refresh_sec,
d.failed_refreshes_last_50,
r.scheduling_state,
CASE
WHEN r.actual_lag_seconds > r.target_lag_seconds * 3 THEN 'CRITICAL'
WHEN r.actual_lag_seconds > r.target_lag_seconds * 1.5 THEN 'WARNING'
ELSE 'OK'
END AS health_status
FROM refresh_stats r
JOIN refresh_durations d ON r.name = d.name
ORDER BY lag_overage_seconds DESC NULLS LAST;
Run this every 30 seconds (via Snowflake Tasks or your orchestrator) and alert when health_status is WARNING or CRITICAL. Pair this with your connector's Kafka consumer group lag metrics — if consumer lag is growing, records aren't reaching Snowflake at all, and no Dynamic Table refresh will help.
Hands-On Exercise
Now let's put this together end-to-end. In this exercise, you'll simulate a Kafka topic using a Python producer, land data in Snowflake via Snowpipe Streaming (using the Python SDK to avoid needing a full Kafka cluster), then create a two-layer Dynamic Table transformation and measure end-to-end latency.
Setup:
pip install snowflake-ingest-sdk==1.1.0 faker
Python Snowpipe Streaming producer (simulates what the Kafka connector does internally):
import time
import json
import uuid
import random
from datetime import datetime, timezone
from faker import Faker
from snowflake.ingest import SimpleIngestManager
from snowflake.ingest.utils.uris import DEFAULT_SCHEME
# Note: In production this is handled by the Kafka connector.
# This exercise uses the Python SDK directly for simplicity.
from snowflake.connector import connect
from snowflake.connector.cursor import DictCursor
fake = Faker()
PAYMENT_METHODS = ['visa_credit', 'mastercard_debit', 'amex_platinum',
'paypal_wallet', 'apple_pay_visa']
STATUSES = ['COMPLETED'] * 7 + ['FAILED'] * 2 + ['PENDING'] * 1 # weighted
def generate_payment_event():
return {
"payment_id": str(uuid.uuid4()),
"customer_id": f"cust_{random.randint(1000, 9999)}",
"merchant_id": f"merch_{random.randint(100, 150)}", # 51 merchants
"amount_cents": random.randint(99, 150000), # $0.99 to $1500
"currency_code": random.choice(["USD", "USD", "USD", "GBP", "EUR"]),
"payment_method": random.choice(PAYMENT_METHODS),
"status": random.choice(STATUSES),
"metadata": {
"device_type": random.choice(["mobile", "web", "pos"]),
"ip_country": fake.country_code()
},
"event_timestamp": datetime.now(timezone.utc).isoformat()
}
def stream_events_to_snowflake(conn, events_per_second=50, duration_seconds=120):
"""
In production, the Kafka connector handles this.
Here we INSERT directly to demonstrate the latency characteristics.
"""
cursor = conn.cursor()
events_sent = 0
start_time = time.time()
print(f"Streaming {events_per_second} events/sec for {duration_seconds}s...")
while time.time() - start_time < duration_seconds:
batch = [generate_payment_event() for _ in range(events_per_second)]
# Build INSERT with multiple rows
values_clause = ", ".join([
f"('{e['payment_id']}', '{e['customer_id']}', '{e['merchant_id']}', "
f"{e['amount_cents']}, '{e['currency_code']}', '{e['payment_method']}', "
f"'{e['status']}', PARSE_JSON('{json.dumps(e['metadata'])}'), "
f"'{e['event_timestamp']}'::TIMESTAMP_TZ, {events_sent + i}, 0, "
f"'payments.events.v1', CURRENT_TIMESTAMP())"
for i, e in enumerate(batch)
])
cursor.execute(f"""
INSERT INTO RAW_DB.KAFKA_INGEST.PAYMENTS_EVENTS_V1
(payment_id, customer_id, merchant_id, amount_cents, currency_code,
payment_method, status, metadata, event_timestamp, kafka_offset,
kafka_partition, kafka_topic, ingested_at)
VALUES {values_clause}
""")
events_sent += len(batch)
print(f"Sent {events_sent} events | Elapsed: {time.time()-start_time:.1f}s")
time.sleep(1)
print(f"Complete. Total events: {events_sent}")
cursor.close()
# Connect and run
conn = connect(
account='yourorg-youraccount',
user='YOUR_USER',
private_key_file='path/to/rsa_key.p8',
warehouse='KAFKA_WH',
role='KAFKA_LOADER_ROLE'
)
stream_events_to_snowflake(conn)
Measure end-to-end latency:
After running the producer for 60 seconds, query the Gold Dynamic Table and measure the gap:
-- How stale is our most recent data?
SELECT
MAX(event_at) AS latest_event_time,
CURRENT_TIMESTAMP() AS query_time,
DATEDIFF('second', MAX(event_at), CURRENT_TIMESTAMP()) AS data_age_seconds,
MAX(gold_refreshed_at) AS last_gold_refresh,
DATEDIFF('second', MAX(gold_refreshed_at), CURRENT_TIMESTAMP()) AS refresh_age_seconds
FROM ANALYTICS_DB.GOLD.PAYMENTS_METRICS_REALTIME;
You should see data_age_seconds in the 15-60 second range depending on your Dynamic Table target lag settings. If it's consistently above 90 seconds, check the refresh history to diagnose where the pipeline is falling behind.
Common Mistakes & Troubleshooting
Mistake 1: Using DOWNSTREAM_REFRESH on Dynamic Tables with Aggregations
When Dynamic Tables reference other Dynamic Tables, Snowflake has two refresh modes: DOWNSTREAM (the default) and INCREMENTAL. For tables with aggregations, INCREMENTAL isn't always possible — Snowflake may fall back to a full recompute silently. Check the refresh_mode and refresh_mode_reason columns in INFORMATION_SCHEMA.DYNAMIC_TABLES. If refresh_mode shows FULL when you expected INCREMENTAL, you're paying the full recompute cost on every refresh. Simplify your SQL to enable incremental mode: remove complex subqueries, replace some CASE expressions with JOINs to lookup tables, and avoid non-deterministic functions like RANDOM().
Mistake 2: Snowpipe Streaming and Transient Table Conflicts
Snowpipe Streaming requires target tables to have CHANGE_TRACKING = TRUE for Dynamic Tables to read from them efficiently. However, if you're using Transient tables (which many teams do in raw layers to avoid storage fees), there's a subtle issue: Transient tables still accumulate micro-partitions from streaming writes, but they don't benefit from Automatic Clustering triggers in the same way as permanent tables. Set your raw landing tables as permanent and enable Automatic Clustering with a monthly cost cap via Resource Monitors.
Mistake 3: Forgetting That dbt `dynamic_table` Refreshes Reset Everything
When you run dbt run on a model with materialized='dynamic_table' and the SQL has changed, dbt executes CREATE OR REPLACE DYNAMIC TABLE, which drops and recreates the table. This means:
- Any Snowflake Streams reading the Dynamic Table are invalidated
- Any downstream Dynamic Tables may require a full recompute on next refresh
- Any Grants on the table are dropped (re-apply via
dbt run with post-hooks)
Always use a deployment strategy that applies DDL changes during maintenance windows, and pair this with Automating dbt Environment Promotion with CI/CD Pipelines, Slim CI, and State-Aware Deployments to control when and how schema changes reach production.
Mistake 4: Not Accounting for Kafka Connector Rebalances
Kafka connector rebalances (triggered by consumer group changes, task restarts, or scaling events) can cause a 30-60 second pause in message delivery. During this window, no new data reaches Snowflake, and your Dynamic Tables will refresh against an unchanged source — reporting stale but not technically erroring. Your monitoring must track Kafka consumer group lag and Dynamic Table lag independently. A healthy Kafka consumer group with high Dynamic Table lag indicates a Snowflake-side issue; a healthy Dynamic Table lag with growing Kafka consumer lag indicates a Kafka-side issue.
Mistake 5: Queries That Break Dynamic Table Incremental Mode
Not all SQL constructs support incremental refreshes in Dynamic Tables. Operations that force full recomputes include: DISTINCT without a GROUP BY, most window functions (except in very specific patterns), self-joins, and lateral flattening of VARIANT arrays in non-trivial ways. Before committing to a Dynamic Table definition, check whether Snowflake resolved it to incremental or full mode using the metadata query above. If it's full mode and your target lag is 30 seconds, every refresh scans your entire Silver table — budget accordingly.
Summary & Next Steps
You now have a complete, production-grade reference architecture for sub-minute analytics freshness using Kafka, Snowflake Snowpipe Streaming, and Dynamic Tables. Let's recap the key design decisions:
- Snowpipe Streaming (not traditional Snowpipe) is the mechanism that makes sub-minute ingestion possible — it bypasses file staging and writes rows directly into Snowflake's buffer
- The lag budget is cumulative: if Bronze → Silver takes 30 seconds and Silver → Gold takes 30 seconds, your end-to-end freshness is ~60 seconds at best. Design target lags as a shared budget across layers
- Dynamic Tables handle the "what" not the "how" of incremental transformation — Snowflake manages the delta computation. Your job is to write SQL that allows incremental mode rather than forcing full recomputes
- dbt complements Dynamic Tables by providing schema management, testing, documentation, and the batch layer, but it doesn't replace them in the streaming path
- Late-arriving data requires a sliding window strategy in your Gold Dynamic Tables, paired with a batch incremental model that corrects the historical record
The architecture you've built here is the streaming ingestion foundation. There are several natural directions to extend it:
- Add data quality alerting using the monitoring queries as a foundation — pipe the health metrics to Slack or PagerDuty via Snowflake Tasks
- Implement lineage tracking across the Kafka → Snowflake → dbt chain using OpenLineage events, which is covered in Multi-Hop Data Lineage Tracking Across the Modern Data Stack: Connecting Ingestion, Transformation, and Consumption with OpenLineage and Marquez
- Secure the pipeline for multi-tenant scenarios — Dynamic Tables inherit Snowflake's RBAC and row-level security policies, so you can apply column masking and row filters at the Gold layer without duplicating data. See Implementing Row-Level Security and Column Masking Policies for Multi-Tenant Analytics in Snowflake and BigQuery for implementation patterns
- Build a semantic layer on top of your Gold Dynamic Tables to give self-serve BI users a governed, metric-consistent view of real-time data
The lines between batch and streaming are blurring. Snowflake Dynamic Tables are one of the clearest expressions of that trend: streaming freshness, batch semantics, warehouse-native execution. Master this architecture and you'll be ahead of most data teams still choosing between the two extremes.
