Cloud Lab: Snowflake CDC with Kafka Connect

Overview

Snowflake is a cloud data warehouse that excels at analytics workloads. Integrating CDC with Snowflake enables near-real-time analytics on operational data. This lab covers two primary approaches:

1. Kafka Connect with Snowflake Sink Connector: Stream CDC events from Kafka topics directly into Snowflake tables
2. Snowpipe: Automatically ingest CDC files from cloud storage (S3, Azure Blob, GCS) as they arrive

Key advantages:

• Zero management of data warehouse infrastructure
• Automatic scaling based on workload
• Native support for semi-structured data (JSON, Avro, Parquet)
• Time Travel for historical queries
• Separation of compute and storage for cost optimization

Prerequisites

• Snowflake Account: Trial account available (free for 30 days with $400 credit)
• Kafka Cluster: Running Debezium or CDC source connector
• Kafka Connect: For Snowflake Sink Connector approach
• Cloud Storage: S3, Azure Blob, or GCS (for Snowpipe approach)
• SnowSQL: Snowflake CLI tool (optional but recommended)

Architecture Options

Option 1: Kafka Connect → Snowflake (Recommended for Kafka users)

┌──────────────┐      ┌──────────────┐      ┌─────────────┐
│   MySQL      │─────▶│  Debezium    │─────▶│   Kafka     │
│  (Source)    │      │  Connector   │      │   Topics    │
└──────────────┘      └──────────────┘      └──────┬──────┘
                                                    │
                                                    │
                      ┌──────────────┐              │
                      │  Snowflake   │◀─────────────┘
                      │    Sink      │
                      │  Connector   │
                      └──────┬───────┘
                             │
                             ▼
                      ┌──────────────┐
                      │  Snowflake   │
                      │   Tables     │
                      └──────────────┘

Option 2: CDC Files → Snowpipe (For file-based CDC)

┌──────────────┐      ┌──────────────┐      ┌─────────────┐
│   MySQL      │─────▶│  AWS DMS /   │─────▶│     S3      │
│  (Source)    │      │  Fivetran    │      │   Bucket    │
└──────────────┘      └──────────────┘      └──────┬──────┘
                                                    │
                                                    │ (Event)
                      ┌──────────────┐              │
                      │  Snowpipe    │◀─────────────┘
                      │  (Auto)      │
                      └──────┬───────┘
                             │
                             ▼
                      ┌──────────────┐
                      │  Snowflake   │
                      │   Tables     │
                      └──────────────┘

Setup: Kafka Connect + Snowflake Sink

Step 1: Create Snowflake Objects

First, set up database, schema, user, and role for CDC ingestion.

-- Connect to Snowflake (SnowSQL or web UI)
USE ROLE ACCOUNTADMIN;

-- Create database and schema
CREATE DATABASE IF NOT EXISTS CDC_DB;
CREATE SCHEMA IF NOT EXISTS CDC_DB.CDC_SCHEMA;

-- Create warehouse for loading
CREATE WAREHOUSE IF NOT EXISTS CDC_WH 
  WITH WAREHOUSE_SIZE = 'XSMALL' 
  AUTO_SUSPEND = 60 
  AUTO_RESUME = TRUE
  INITIALLY_SUSPENDED = TRUE;

-- Create role and user for Kafka Connect
CREATE ROLE IF NOT EXISTS CDC_ROLE;
GRANT USAGE ON WAREHOUSE CDC_WH TO ROLE CDC_ROLE;
GRANT USAGE ON DATABASE CDC_DB TO ROLE CDC_ROLE;
GRANT USAGE ON SCHEMA CDC_DB.CDC_SCHEMA TO ROLE CDC_ROLE;
GRANT CREATE TABLE ON SCHEMA CDC_DB.CDC_SCHEMA TO ROLE CDC_ROLE;
GRANT CREATE STAGE ON SCHEMA CDC_DB.CDC_SCHEMA TO ROLE CDC_ROLE;
GRANT CREATE PIPE ON SCHEMA CDC_DB.CDC_SCHEMA TO ROLE CDC_ROLE;

-- Create user with key-pair authentication (recommended)
CREATE USER IF NOT EXISTS kafka_connector_user
  DEFAULT_ROLE = CDC_ROLE
  DEFAULT_WAREHOUSE = CDC_WH
  DEFAULT_NAMESPACE = CDC_DB.CDC_SCHEMA
  COMMENT = 'User for Kafka Connect Snowflake Sink';

GRANT ROLE CDC_ROLE TO USER kafka_connector_user;

Step 2: Generate Key-Pair for Authentication

For production, use key-pair authentication instead of passwords.

# Generate private key
openssl genrsa 2048 | openssl pkcs8 -topk8 -inform PEM -out rsa_key.p8 -nocrypt

# Generate public key
openssl rsa -in rsa_key.p8 -pubout -out rsa_key.pub

# Get public key value (remove header/footer, join lines)
cat rsa_key.pub | grep -v "BEGIN PUBLIC" | grep -v "END PUBLIC" | tr -d '\n'

# Assign public key to Snowflake user
ALTER USER kafka_connector_user SET RSA_PUBLIC_KEY='MIIBIjANBg...';

Step 3: Install Snowflake Kafka Connector

Use the 2.x line. The 1.9.x releases (circa 2022) are superseded and predate Snowpipe Streaming.

# Download the current 2.x connector from Snowflake
wget https://repo1.maven.org/maven2/com/snowflake/snowflake-kafka-connector/2.5.0/snowflake-kafka-connector-2.5.0.jar

# Copy to Kafka Connect plugins directory
mkdir -p /opt/kafka/plugins/snowflake-connector
cp snowflake-kafka-connector-2.5.0.jar /opt/kafka/plugins/snowflake-connector/

# Restart Kafka Connect to load plugin
docker-compose restart connect
# Or: systemctl restart kafka-connect

Step 4: Configure Snowflake Sink Connector

Create snowflake-sink-config.json:

{
  "name": "snowflake-sink-connector",
  "config": {
    "connector.class": "com.snowflake.kafka.connector.SnowflakeSinkConnector",
    "tasks.max": "2",
    
    "topics": "dbserver1.inventory.customers,dbserver1.inventory.orders",
    
    "snowflake.url.name": "https://myaccount.snowflakecomputing.com:443",
    "snowflake.user.name": "kafka_connector_user",
    "snowflake.private.key": "MIIEvQIBADANBgkqhkiG...",
    "snowflake.database.name": "CDC_DB",
    "snowflake.schema.name": "CDC_SCHEMA",
    
    "buffer.count.records": "10000",
    "buffer.flush.time": "60",
    "buffer.size.bytes": "10000000",
    
    "snowflake.topic2table.map": "dbserver1.inventory.customers:CUSTOMERS,dbserver1.inventory.orders:ORDERS",
    
    "key.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "key.converter.schemas.enable": "false",
    "value.converter.schemas.enable": "true",
    
    "snowflake.metadata.createtime": "true",
    "snowflake.metadata.topic": "true",
    "snowflake.metadata.offset.and.partition": "true",
    
    "errors.tolerance": "all",
    "errors.log.enable": "true",
    "errors.deadletterqueue.topic.name": "dlq.snowflake",
    "errors.deadletterqueue.context.headers.enable": "true"
  }
}

Step 5: Deploy the Connector

# Deploy via REST API
curl -i -X POST -H "Accept:application/json" -H "Content-Type:application/json" \
  http://localhost:8083/connectors/ -d @snowflake-sink-config.json

# Check connector status
curl http://localhost:8083/connectors/snowflake-sink-connector/status | jq

# View connector tasks
curl http://localhost:8083/connectors/snowflake-sink-connector/tasks | jq

Step 6: Verify Data in Snowflake

-- Check that tables were created
USE SCHEMA CDC_DB.CDC_SCHEMA;
SHOW TABLES;

-- Query data from customers table
SELECT * FROM CUSTOMERS LIMIT 10;

-- Check metadata columns.
-- The Kafka connector lands RECORD_METADATA as a VARIANT — access sub-fields
-- with the ':' path operator, NOT the '$' pseudo-columns (RECORD_METADATA$...).
-- ($FILENAME / $FILE_ROW_NUMBER are Snowpipe COPY file metadata and do not
--  exist on the Kafka-connector path.)
SELECT
  RECORD_METADATA:topic::STRING      as topic,
  RECORD_METADATA:partition::INT     as kafka_partition,
  RECORD_METADATA:offset::BIGINT     as kafka_offset,
  RECORD_METADATA:CreateTime::BIGINT as create_time_ms,
  RECORD_CONTENT
FROM CUSTOMERS;

-- Parse JSON content (Snowflake stores Kafka messages as VARIANT)
SELECT 
  RECORD_CONTENT:payload:after:id::INT as customer_id,
  RECORD_CONTENT:payload:after:name::STRING as customer_name,
  RECORD_CONTENT:payload:after:email::STRING as email,
  RECORD_CONTENT:payload:op::STRING as operation
FROM CUSTOMERS
WHERE RECORD_CONTENT:payload:op::STRING = 'c' -- inserts only
LIMIT 10;

Setup: Snowpipe for File-Based CDC

Step 1: Create External Stage

Configure Snowflake to access your S3 bucket with CDC files.

-- Create storage integration (one-time setup)
USE ROLE ACCOUNTADMIN;

CREATE STORAGE INTEGRATION s3_cdc_integration
  TYPE = EXTERNAL_STAGE
  STORAGE_PROVIDER = S3
  ENABLED = TRUE
  STORAGE_AWS_ROLE_ARN = 'arn:aws:iam::123456789012:role/snowflake-s3-role'
  STORAGE_ALLOWED_LOCATIONS = ('s3://my-cdc-bucket/cdc-data/');

-- Get AWS IAM user for Snowflake (for trust policy)
DESC STORAGE INTEGRATION s3_cdc_integration;
-- Note STORAGE_AWS_IAM_USER_ARN and STORAGE_AWS_EXTERNAL_ID

-- Create external stage
USE SCHEMA CDC_DB.CDC_SCHEMA;

CREATE STAGE s3_cdc_stage
  URL = 's3://my-cdc-bucket/cdc-data/'
  STORAGE_INTEGRATION = s3_cdc_integration
  FILE_FORMAT = (TYPE = 'PARQUET');

-- Test stage connection
LIST @s3_cdc_stage;

Step 2: Configure AWS IAM Trust Policy

Allow Snowflake to assume role to read S3 bucket:

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Principal": {
        "AWS": "arn:aws:iam::123456789012:user/abc12345-s"
      },
      "Action": "sts:AssumeRole",
      "Condition": {
        "StringEquals": {
          "sts:ExternalId": "ABC123_SFCRole=1_xyz789=="
        }
      }
    }
  ]
}

Step 3: Create Target Table

CREATE TABLE customers_cdc (
  id INTEGER,
  name STRING,
  email STRING,
  created_at TIMESTAMP,
  cdc_operation STRING,
  cdc_timestamp TIMESTAMP,
  -- Metadata columns
  load_timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP(),
  source_file STRING
);

Step 4: Create Snowpipe

Single-pipe path: leave AUTO_INGEST = TRUE and let Snowflake provision its own SQS queue for the pipe. You wire S3 directly to that queue in Step 5 (no SNS topic in between). Do not also set AWS_SNS_TOPIC — that is the SNS fan-out pattern (one topic notifying many pipes) and it will not fire if S3 is pointed at the pipe's SQS queue instead.

CREATE PIPE customers_cdc_pipe
  AUTO_INGEST = TRUE
AS
COPY INTO customers_cdc (id, name, email, created_at, cdc_operation, cdc_timestamp, source_file)
FROM (
  SELECT
    $1:id::INTEGER,
    $1:name::STRING,
    $1:email::STRING,
    $1:created_at::TIMESTAMP,
    $1:Op::STRING,
    $1:cdc_timestamp::TIMESTAMP,
    METADATA$FILENAME
  FROM @s3_cdc_stage
)
FILE_FORMAT = (TYPE = 'PARQUET')
ON_ERROR = 'CONTINUE';

-- Get the pipe's own SQS notification channel for the S3 event config (Step 5).
-- Copy the notification_channel ARN from this output into s3-notification.json.
SHOW PIPES;
SELECT SYSTEM$PIPE_STATUS('customers_cdc_pipe');

-- Rows Snowpipe skipped under ON_ERROR = 'CONTINUE'
SELECT FILE_NAME, ROW_COUNT, ERROR_COUNT, FIRST_ERROR_MESSAGE, LAST_LOAD_TIME
FROM TABLE(INFORMATION_SCHEMA.COPY_HISTORY(
  TABLE_NAME  => 'CUSTOMERS_CDC',
  START_TIME  => DATEADD(hour, -24, CURRENT_TIMESTAMP())
))
WHERE ERROR_COUNT > 0
ORDER BY LAST_LOAD_TIME DESC;

Step 5: Configure S3 Event Notifications

Point S3 straight at the pipe's own SQS queue. The QueueArn below is the notification_channel value from SHOW PIPES in Step 4 — S3 → SQS → Snowpipe, one hop, no SNS topic.

# AWS CLI: Configure S3 bucket notification
aws s3api put-bucket-notification-configuration \
  --bucket my-cdc-bucket \
  --notification-configuration file://s3-notification.json

# s3-notification.json
# QueueArn = the pipe's notification_channel (from SHOW PIPES, Step 4)
{
  "QueueConfigurations": [
    {
      "Id": "snowpipe-cdc-events",
      "QueueArn": "arn:aws:sqs:us-east-1:123456789012:sf-snowpipe-ABCD1234",
      "Events": ["s3:ObjectCreated:*"],
      "Filter": {
        "Key": {
          "FilterRules": [
            {
              "Name": "prefix",
              "Value": "cdc-data/"
            },
            {
              "Name": "suffix",
              "Value": ".parquet"
            }
          ]
        }
      }
    }
  ]
}

Step 6: Test Snowpipe

-- Upload test file to S3 to trigger ingestion
-- Then check pipe status
SELECT SYSTEM$PIPE_STATUS('customers_cdc_pipe');

-- View load history
SELECT * FROM TABLE(INFORMATION_SCHEMA.COPY_HISTORY(
  TABLE_NAME => 'CUSTOMERS_CDC',
  START_TIME => DATEADD(hours, -1, CURRENT_TIMESTAMP())
));

-- Query loaded data
SELECT COUNT(*) FROM customers_cdc;
SELECT * FROM customers_cdc ORDER BY load_timestamp DESC LIMIT 10;

Materializing CDC Events

Raw CDC events contain operations (INSERT, UPDATE, DELETE). To build a current state table, apply MERGE logic:

-- Create current state table
CREATE TABLE customers_current (
  id INTEGER PRIMARY KEY,
  name STRING,
  email STRING,
  created_at TIMESTAMP,
  updated_at TIMESTAMP
);

-- Merge CDC events (run periodically or via task).
-- Dedup ("latest wins") MUST order by log position, never by ts_ms.
MERGE INTO customers_current AS target
USING (
  SELECT
    RECORD_CONTENT:payload:after:id::INT as id,
    RECORD_CONTENT:payload:after:name::STRING as name,
    RECORD_CONTENT:payload:after:email::STRING as email,
    RECORD_CONTENT:payload:after:created_at::TIMESTAMP as created_at,
    -- ts_ms is epoch-MILLISECONDS; TO_TIMESTAMP_NTZ(..., 3) reads scale-3.
    -- A bare ::TIMESTAMP cast treats it as seconds and lands in the year ~55000.
    TO_TIMESTAMP_NTZ(RECORD_CONTENT:payload:ts_ms::NUMBER, 3) as updated_at, -- metric only, NOT for ordering
    RECORD_CONTENT:payload:op::STRING as op
  FROM CUSTOMERS
  -- Last write wins by LOG POSITION. The authoritative, globally-monotonic
  -- order is the source DB commit position from the Debezium envelope
  -- (Postgres LSN below; MySQL source:file+pos; Oracle source:scn) — correct
  -- across partitions, and the ordering this site recommends.
  QUALIFY ROW_NUMBER() OVER (
    PARTITION BY id
    ORDER BY RECORD_CONTENT:payload:source:lsn::BIGINT DESC
    --   MySQL:  RECORD_CONTENT:payload:source:file::STRING DESC,
    --           RECORD_CONTENT:payload:source:pos::BIGINT  DESC
    --   Oracle: RECORD_CONTENT:payload:source:scn::BIGINT  DESC
  ) = 1
  -- A Kafka offset (RECORD_METADATA:offset DESC) is valid ONLY if the topic is
  -- keyed on the PK so each id stays in one partition — offsets are not
  -- comparable across partitions. Prefer the source position above.
) AS source
ON target.id = source.id
WHEN MATCHED AND source.op = 'd' THEN DELETE
WHEN MATCHED THEN UPDATE SET
  target.name = source.name,
  target.email = source.email,
  target.updated_at = source.updated_at
WHEN NOT MATCHED AND source.op != 'd' THEN INSERT (
  id, name, email, created_at, updated_at
) VALUES (
  source.id, source.name, source.email, source.created_at, source.updated_at
);

Automate with Snowflake Tasks

-- Create task to materialize every 5 minutes
CREATE TASK materialize_customers_cdc
  WAREHOUSE = CDC_WH
  SCHEDULE = '5 MINUTE'
AS
MERGE INTO customers_current AS target
USING (
  -- Same MERGE logic as above — dedup by LOG POSITION, never ts_ms.
  SELECT
    RECORD_CONTENT:payload:after:id::INT as id,
    RECORD_CONTENT:payload:after:name::STRING as name,
    RECORD_CONTENT:payload:after:email::STRING as email,
    RECORD_CONTENT:payload:after:created_at::TIMESTAMP as created_at,
    TO_TIMESTAMP_NTZ(RECORD_CONTENT:payload:ts_ms::NUMBER, 3) as updated_at, -- epoch-ms; metric only
    RECORD_CONTENT:payload:op::STRING as op
  FROM CUSTOMERS
  QUALIFY ROW_NUMBER() OVER (
    PARTITION BY id
    -- dedup by source commit position (LSN / file+pos / SCN), never ts_ms or a
    -- cross-partition Kafka offset — see the detailed MERGE above.
    ORDER BY RECORD_CONTENT:payload:source:lsn::BIGINT DESC
  ) = 1
) AS source
ON target.id = source.id
WHEN MATCHED AND source.op = 'd' THEN DELETE
WHEN MATCHED THEN UPDATE SET
  target.name = source.name,
  target.email = source.email,
  target.updated_at = source.updated_at
WHEN NOT MATCHED AND source.op != 'd' THEN INSERT (
  id, name, email, created_at, updated_at
) VALUES (
  source.id, source.name, source.email, source.created_at, source.updated_at
);

-- Enable task
ALTER TASK materialize_customers_cdc RESUME;

-- Monitor task history
SELECT * FROM TABLE(INFORMATION_SCHEMA.TASK_HISTORY(
  TASK_NAME => 'MATERIALIZE_CUSTOMERS_CDC',
  SCHEDULED_TIME_RANGE_START => DATEADD(hour, -24, CURRENT_TIMESTAMP())
));

Snowflake-Specific Gotchas

-- Add second public key (key rotation grace period)
ALTER USER kafka_connector_user SET RSA_PUBLIC_KEY_2='NEW_KEY...';

-- After connector updated, remove old key
ALTER USER kafka_connector_user UNSET RSA_PUBLIC_KEY;

SELECT * FROM SNOWFLAKE.ACCOUNT_USAGE.PIPE_USAGE_HISTORY
WHERE PIPE_NAME = 'CUSTOMERS_CDC_PIPE'
ORDER BY START_TIME DESC;

Monitoring & Observability

Key Metrics to Track

-- Lag: Time between source change and Snowflake load
SELECT 
  TABLE_NAME,
  DATEDIFF(second, 
    MAX(RECORD_CONTENT:payload:ts_ms)::TIMESTAMP,
    CURRENT_TIMESTAMP()
  ) as lag_seconds
FROM CUSTOMERS
GROUP BY TABLE_NAME;

-- Ingestion rate (rows per minute)
SELECT 
  DATE_TRUNC('minute', load_timestamp) as minute,
  COUNT(*) as rows_ingested
FROM customers_cdc
WHERE load_timestamp > DATEADD(hour, -1, CURRENT_TIMESTAMP())
GROUP BY DATE_TRUNC('minute', load_timestamp)
ORDER BY minute DESC;

-- Error rate from COPY_HISTORY
SELECT 
  TABLE_NAME,
  STATUS,
  COUNT(*) as load_count,
  SUM(ROW_COUNT) as total_rows,
  SUM(ERROR_COUNT) as total_errors
FROM INFORMATION_SCHEMA.COPY_HISTORY
WHERE LAST_LOAD_TIME > DATEADD(day, -1, CURRENT_TIMESTAMP())
GROUP BY TABLE_NAME, STATUS;

Alerting with Snowflake Alerts

-- Create alert for high CDC lag
CREATE ALERT cdc_lag_alert
  WAREHOUSE = CDC_WH
  SCHEDULE = '5 MINUTE'
  IF (EXISTS (
    SELECT 1
    FROM CUSTOMERS
    WHERE DATEDIFF(minute, 
      MAX(RECORD_CONTENT:payload:ts_ms)::TIMESTAMP,
      CURRENT_TIMESTAMP()
    ) > 30
  ))
THEN
  CALL SYSTEM$SEND_EMAIL(
    'NOTIFICATION_INTEGRATION',
    'ops@example.com',
    'CDC Lag Alert',
    'CDC lag exceeded 30 minutes'
  );

ALTER ALERT cdc_lag_alert RESUME;

Cost Optimization

• Warehouse Size: Start with XSMALL ($2/hour, $0.0006/second). Scale up only if needed.
• Auto-Suspend: Set to 60 seconds for CDC workloads to minimize idle time.
• Clustering: Add cluster keys on frequently filtered columns (e.g., timestamp) to reduce scan cost.
• Time Travel: Reduce retention to 1 day for non-critical CDC tables (default: 1 day, max: 90 days Enterprise).
• Snowpipe: More cost-effective than warehouse for continuous micro-batch loads.

-- Set Time Travel retention
ALTER TABLE customers_cdc SET DATA_RETENTION_TIME_IN_DAYS = 1;

-- Add clustering for query performance
ALTER TABLE customers_cdc CLUSTER BY (DATE_TRUNC('day', load_timestamp));

Snowflake vs. Traditional Data Warehouse