Overview

Demo: Application Log Streaming

{
  "timestamp": "2025-10-15T10:23:45.123Z",
  "level": "INFO",
  "service": "web-api",
  "host": "api-server-01",
  "message": "Request processed successfully",
  "request_id": "req-abc123",
  "duration_ms": 45,
  "status_code": 200
}

What is Openflow

About the Openflow Kafka Connector

Advanced Option: For production scenarios requiring Kafka metadata (partition, offset), dead letter queues, or Iceberg table support, see the DLQ and metadata connector.

What You Will Learn

What You Will Build

Prerequisites

Don't have Kafka yet? We tested this with Confluent Cloud, which has a free tier perfect for this quickstart. You can spin up a cluster in about 5 minutes. Other managed options like AWS MSK or GCP Managed Kafka work great too.

Using Openflow BYOC? This quickstart focuses on SPCS deployment (Snowflake-managed), but the connector setup and concepts are identical for BYOC deployments—only the runtime deployment steps differ.

Clone Repository

git clone https://github.com/Snowflake-Labs/sfguide-getting-started-openflow-kafka-connector.git
cd sfguide-getting-started-openflow-kafka-connector
export QUICK_START_REPO=$PWD

What's in the Repository?

Prefer Notebooks? If you're more comfortable with Jupyter notebooks, you can follow along using our interactive notebook version instead. It contains the same content in a notebook-friendly format.

Quick Tip: The generate_logs.py script has a --test-connection flag that's super helpful for verifying your Kafka setup works before we start. We'll use this later!

Configure Kafka

Kafka Requirements

Required Kafka Configuration:

• Kafka cluster available: You have a running Kafka cluster (GCP Managed Kafka, AWS MSK, Confluent Cloud, or self-hosted) with network accessibility from Snowflake.
• Topic creation permissions: Ability to create topics or use existing topics
• Producer access: Ability to produce messages to topics (for testing)
• Broker endpoints: Note down your Kafka broker addresses (e.g., broker1:9092,broker2:9092)
• Security protocol: Know your security configuration (SASL_SSL)
• SASL mechanism: Know your SASL mechanism (PLAIN)

This quickstart was tested with Confluent Cloud.

Setup Redpanda CLI (rpk)

brew install redpanda-data/tap/redpanda

curl -1sLf 'https://dl.redpanda.com/nzc4ZYQK3WRGd9sy/redpanda/cfg/setup/bash.deb.sh' | sudo -E bash
sudo apt install redpanda -y

curl -1sLf 'https://dl.redpanda.com/nzc4ZYQK3WRGd9sy/redpanda/cfg/setup/bash.rpm.sh' | sudo -E bash
sudo yum install redpanda -y

rpk version

Alternative: If you prefer traditional Kafka CLI tools, they work too. Just replace rpk commands with equivalent kafka-topics.sh commands. However, we recommend rpk for its improved usability.

Configure rpk Profile

Important - Kafka Credentials Required: Before proceeding, ensure you have obtained the following from your Kafka cluster administrator or cloud console:

• Kafka Broker Address(es): Public endpoint(s) of your Kafka brokers (e.g., broker-1.example.com:9092)
• Username: Your SASL authentication username (e.g. "api-key" in Confluent Cloud)
• Password: Your SASL authentication password (e.g. "api-secret" in Confluent Cloud)

For this quickstart, we assume:

• Security Protocol: SASL_SSL (encrypted connection with authentication)
• SASL Mechanism: PLAIN (username/password authentication)

# Create a profile for this quickstart
rpk profile create quickstart-kafka

# Set broker and security settings
# Replace YOUR-KAFKA-BROKER, YOUR-USER-NAME, and YOUR-PASSWORD with your actual Kafka broker, username, and password
rpk profile set kafka_api.tls.enabled="true"
rpk profile set brokers="YOUR-KAFKA-BROKER:9092"
rpk profile set sasl.mechanism="PLAIN"
rpk profile set user="YOUR-USER-NAME"
rpk profile set pass="YOUR-PASSWORD"

Confluent Cloud Users: Use your Confluent Cloud API Key as the username and API Secret as the password. These can be created in the Confluent Cloud Console under "Data Integration" → "API Keys".

# View current profile
rpk profile print

# Test connection
rpk cluster info

Profile Benefits: With a configured profile, you don't need to specify --brokers or authentication details in subsequent rpk commands. The profile stores all connection settings securely.

Kafka Service Options

Tested Environment: This quickstart was tested with Confluent Cloud using SASL_SSL security protocol and PLAIN authentication mechanism.

Create Kafka Topic

# Create topic with 3 partitions for scalability
rpk topic create application-logs \
  --partitions 3 \
  --replicas 3

TOPIC            STATUS
application-logs  OK

Topic Naming: You can use any topic name you prefer. If you use a different name, make sure to update the connector configuration accordingly. The default setup assumes application-logs.

Verify Kafka Cluster and Topic

# List all topics
rpk topic list

# Describe the topic details
rpk topic describe application-logs

Record Cluster Information

# Display cluster information
rpk cluster info

CLUSTER
=======
lkc-ddwwmo

BROKERS
=======
ID    HOST                          PORT
0     broker-1.example.com          9092
1     broker-2.example.com          9092
2     broker-3.example.com          9092
...

TOPICS
======
NAME              PARTITIONS  REPLICAS
application-logs  3           3
topic_0           6           3
...

Setup Log Generator

# Install uv (if not already installed)
curl -LsSf https://astral.sh/uv/install.sh | sh

# Create virtual environment and install dependencies
uv venv
source .venv/bin/activate  # On Windows: .venv\Scripts\activate
uv sync

pip install kafka-python python-dotenv

# Copy template to .env
cp $QUICK_START_REPO/.env.template .env

# Edit .env with your Kafka credentials
# Update the following values:
#   KAFKA_BOOTSTRAP_SERVERS - Your Kafka broker endpoint(s)
#   KAFKA_SASL_USERNAME - Your API Key/username
#   KAFKA_SASL_PASSWORD - Your API Secret/password

Required Configuration: The .env file is required for the generate_logs.py script to work. Update KAFKA_BOOTSTRAP_SERVERS, KAFKA_SASL_USERNAME, and KAFKA_SASL_PASSWORD with your actual Kafka credentials from the rpk profile setup. The template already has KAFKA_SECURITY_PROTOCOL=SASL_SSL and KAFKA_SASL_MECHANISM=PLAIN set correctly.

python $QUICK_START_REPO/sample-data/generate_logs.py --test-connection

============================================================
KAFKA CONNECTION TEST
============================================================

[1/4] Testing connection to Kafka brokers...
      Brokers: pkc-n98pk.us-west-2.aws.confluent.cloud:9092
      Security: SASL_SSL
      SASL Mechanism: PLAIN
      SASL Username: FNEI67I3U52IIGG5
      ✓ Successfully connected to Kafka brokers

[2/4] Fetching cluster metadata...
      ✓ Connected to cluster

[3/4] Checking topic accessibility...
      Topic: application-logs
      ✓ Topic 'application-logs' exists and is accessible
      ✓ Topic has 3 partition(s): [0, 1, 2]

[4/4] Testing write permissions...
      Sending test message to topic 'application-logs'...
      ✓ Successfully sent test message
      ✓ Written to partition 2 at offset 6

============================================================
CONNECTION TEST RESULT: ✓ SUCCESS
============================================================

Your Kafka configuration is working correctly!
You can now produce logs with:
  python generate_logs.py --count 100

Troubleshooting: If the connection test fails, verify:

• Your .env file has correct values matching your rpk profile
• The broker endpoints are accessible from your network
• Your API Key and API Secret are correct
• Security protocol and SASL mechanism match your cluster's configuration

Next Steps

Configure Snowflake

Run Snowflake Setup Script

Customize Network Rule

IMPORTANT - Get Your Kafka Broker List: Before customizing the network rule, extract your broker endpoints from the Kafka cluster info you recorded earlier.

Quick Extract Command (if you have jq installed):

rpk cluster info --format=json | jq -r '[.brokers[].host] | map("'"'"'" + . + "'"'"'") | join(",")'

Example output: 'broker-1.example.com','broker-2.example.com','broker-3.example.com'

This formatted list is ready to paste into the network rule's VALUE_LIST.

-- IMPORTANT: Update the network rule with your Kafka broker endpoints
CREATE OR REPLACE NETWORK RULE QUICKSTART_KAFKA_CONNECTOR_DB.NETWORKS.kafka_network_rule
  MODE = EGRESS
  TYPE = HOST_PORT
  VALUE_LIST = (
    'YOUR-KAFKA-BROKER-1:9092',  -- Replace with your actual Kafka broker
    'YOUR-KAFKA-BROKER-2:9092',  -- Add additional brokers if available
    'YOUR-KAFKA-BROKER-3:9092'   -- Remove if you have fewer brokers
  );

Multiple Brokers

VALUE_LIST = (
  'broker-1.kafka.example.com:9092',
  'broker-2.kafka.example.com:9092',
  'broker-3.kafka.example.com:9092'
)

Using Wildcards for Similar Domains

VALUE_LIST = ('*.kafka.example.com:9092')

Execute the Setup Script

-- Verify objects
USE ROLE QUICKSTART_ROLE;
SHOW DATABASES LIKE 'QUICKSTART_KAFKA_CONNECTOR_DB';
SHOW SCHEMAS IN DATABASE QUICKSTART_KAFKA_CONNECTOR_DB;
SHOW WAREHOUSES LIKE 'QUICKSTART_KAFKA_CONNECTOR_WH';
SHOW NETWORK RULES IN SCHEMA QUICKSTART_KAFKA_CONNECTOR_DB.NETWORKS;
SHOW EXTERNAL ACCESS INTEGRATIONS LIKE 'quickstart_kafka_connector_access';

Setup Openflow Runtime

New to Openflow Snowflake Deployments? For more comprehensive coverage, see the dedicated Getting Started with Openflow Snowflake Deployments quickstart. The steps below provide everything you need to get started quickly.

Create Openflow Deployment

Create Openflow Runtime

Important: Make sure to attach the quickstart_kafka_connector_access external access integration when creating the runtime. This allows the runtime to connect to your Kafka brokers through Snowflake's network rules.

Add Kafka Connector

New Tab & Authentication: Adding the connector to your runtime will open a new runtime tab in your browser with the canvas showing the Openflow Kafka; JSON + SASL + Schema evolution (abbreviated as Kafka JSON SASL SCHEMAEV) processor group. If prompted, authenticate/authorize access to allow the canvas to access the runtime. You'll configure this connector in the next section.

Configure Kafka Connector

About the Connector Name: "Kafka JSON SASL SCHEMAEV" is the abbreviated form of Openflow Kafka; JSON + SASL + Schema evolution. This name indicates the connector supports JSON data format, SASL authentication, and automatic schema evolution.

Configure Kafka Source Parameters

Access Kafka Source Parameters

Connection Details: These values should match the credentials you configured in the rpk profile during the Setup Kafka Environment section. If you're using Confluent Cloud, these are your API Key (username) and API Secret (password).

Configure Kafka Destination Parameters

IMPORTANT: Ensure you have completed the Setup Snowflake Environment section before proceeding. The Kafka connector requires the database, warehouse, and role to be created in advance.

Access Kafka Destination Parameters

Session Token Authentication: When using SNOWFLAKE_SESSION_TOKEN, the connector authenticates using the Openflow runtime's Snowflake session. This is the recommended approach for SPCS deployments as it eliminates the need to manage separate credentials.

Configure Kafka Ingestion Parameters

Access Kafka Ingestion Parameters

Configuration Notes:

• Auto Offset Reset: Use latest for real-time streaming (only new messages), or earliest to replay all existing messages in the topic.
• Consumer Group: The group ID allows Kafka to track which messages have been consumed. If you restart the connector, it will resume from where it left off.
• Table Naming: By default, Openflow creates tables matching your topic names exactly but converted to uppercase with double quotes following Snowflake standards (e.g., topic application-logs → table "APPLICATION-LOGS").

Navigate Back to Process Group

Start the Connector

Start Process Group

Note: You may see an error message about "Snowflake Private Key Service" failing to enable. This error can be safely ignored since this quickstart uses SNOWFLAKE_SESSION_TOKEN authentication instead of private key authentication. The connector will function normally with session token authentication.

What's Happening? The connector:

1. Connects to Kafka: Establishes a connection to your Kafka brokers using the configured credentials
2. Subscribes to Topics: Begins consuming messages from the application-logs topic
3. Streams to Snowflake: Writes records to Snowflake in near real-time using Snowpipe Streaming
4. Tracks Offsets: Maintains consumer group offsets so it can resume from where it left off if restarted

Schema Evolution

Understanding Automatic Schema Detection

Key Benefit: As your application evolves and adds new fields to log messages, corresponding columns are added automatically—no pipeline reconfiguration or manual ALTER TABLE statements needed!

---

Generate Base Logs

Install Dependencies (If Not Already Done)

# Install kafka-python library (skip if already installed)
pip install kafka-python python-dotenv

# Or if you prefer conda
conda install -c conda-forge kafka-python python-dotenv

Produce Base Schema Logs

Using Python Generator (Recommended)

# Produce base logs (WITHOUT evolved fields)
python $QUICK_START_REPO/sample-data/generate_logs.py --count 50

Connecting to Kafka brokers: pkc-xxxxx.aws.confluent.cloud:9092
✓ Connected to Kafka
Producing 50 log events (base schema) to topic 'application-logs'...
  Sent 10/50 events (partition: 0, offset: 125)
  Sent 20/50 events (partition: 1, offset: 89)
  ...
✓ Successfully produced 50 log events (base schema)

About Record Counts: Both methods (Python generator and sample file) produce exactly 50 base schema records for consistency.

• If you ran --test-connection earlier, you'll have 51 total records
• Multiple runs will accumulate records (not replace them)
• All records include the same 11 base schema fields (some with NULL values where not applicable)

Using rpk with Sample File

rpk topic produce application-logs -f '%v{json}\n' < $QUICK_START_REPO/sample-data/sample_logs.json

Generate Custom Sample Files

# Generate custom base schema file
python $QUICK_START_REPO/sample-data/generate_logs.py --count 100 --output my_logs.json

# Generate custom evolved schema file
python $QUICK_START_REPO/sample-data/generate_logs.py --count 50 --evolved --output my_evolved_logs.json

Verify Table Creation

-- Set context
USE ROLE QUICKSTART_ROLE;
USE DATABASE QUICKSTART_KAFKA_CONNECTOR_DB;
USE SCHEMA PUBLIC;
USE WAREHOUSE QUICKSTART_KAFKA_CONNECTOR_WH;

-- Check that table was created
SHOW TABLES LIKE 'APPLICATION-LOGS';

-- Count records
SELECT COUNT(*) as TOTAL_RECORDS FROM "APPLICATION-LOGS";
-- Expected: ~50 records (or 51 if you ran --test-connection earlier)

📸 Capture Initial Schema (Important!)

-- Show initial schema columns with data types (sorted alphabetically)
SELECT 
  COLUMN_NAME, 
  DATA_TYPE
FROM QUICKSTART_KAFKA_CONNECTOR_DB.INFORMATION_SCHEMA.COLUMNS 
WHERE TABLE_NAME ILIKE 'APPLICATION-LOGS'
ORDER BY COLUMN_NAME;

AMOUNT          NUMBER
DURATION_MS     NUMBER
ERROR           TEXT
HOST            TEXT
LEVEL           TEXT
MESSAGE         TEXT
REQUEST_ID      TEXT
SERVICE         TEXT
STATUS_CODE     NUMBER
TIMESTAMP       TEXT
USER_ID         TEXT

Remember These 11 Columns: We'll run this query again in Phase 2 and see many NEW columns appear automatically!

Note: Some columns like AMOUNT, USER_ID, and ERROR may contain NULL values for records where those fields don't apply. For example, only payment-related logs will have AMOUNT values, and only ERROR-level logs will have the ERROR field populated.

Query Base Schema Data

-- View sample records
SELECT 
  TIMESTAMP,
  LEVEL,
  SERVICE,
  MESSAGE,
  STATUS_CODE,
  DURATION_MS
FROM "APPLICATION-LOGS"
ORDER BY TIMESTAMP DESC
LIMIT 10;

-- Log level distribution
SELECT 
  LEVEL,
  COUNT(*) as LOG_COUNT
FROM "APPLICATION-LOGS"
GROUP BY LEVEL
ORDER BY LOG_COUNT DESC;

-- Error analysis (ERROR field only appears in ERROR logs)
SELECT 
  TIMESTAMP,
  SERVICE,
  MESSAGE,
  ERROR,
  STATUS_CODE
FROM "APPLICATION-LOGS"
WHERE ERROR IS NOT NULL
ORDER BY TIMESTAMP DESC
LIMIT 10;

-- Service health summary
SELECT 
  SERVICE,
  COUNT(*) as TOTAL_LOGS,
  SUM(CASE WHEN LEVEL = 'ERROR' THEN 1 ELSE 0 END) as ERROR_COUNT,
  SUM(CASE WHEN LEVEL = 'WARN' THEN 1 ELSE 0 END) as WARN_COUNT,
  ROUND(AVG(DURATION_MS), 2) as AVG_DURATION_MS
FROM "APPLICATION-LOGS"
GROUP BY SERVICE
ORDER BY ERROR_COUNT DESC;

Full Verification Script: Run sql/2a.verify_base_schema.sql for comprehensive Phase 1 verification queries including performance analysis, error summaries, and payment transactions.

---

Generate Evolved Logs

Produce Evolved Schema Logs

Using Python Generator with --evolved Flag (Recommended)

python $QUICK_START_REPO/sample-data/generate_logs.py --count 80 --evolved

Producing 80 log events (evolved schema) to topic 'application-logs'...
  Sent 10/80 events (partition: 0, offset: 175)
  Sent 20/80 events (partition: 1, offset: 89)
  ...
  Sent 80/80 events (partition: 2, offset: 201)
✓ Successfully produced 80 log events (evolved schema)

Using rpk with Evolved Sample File

rpk topic produce application-logs -f '%v{json}\n' < $QUICK_START_REPO/sample-data/sample_logs_evolved.json

🎯 Compare Schemas - NEW Columns Appeared

-- Show columns NOW - compare with Phase 1!
SELECT 
  COLUMN_NAME, 
  DATA_TYPE
FROM QUICKSTART_KAFKA_CONNECTOR_DB.INFORMATION_SCHEMA.COLUMNS 
WHERE TABLE_NAME ILIKE 'APPLICATION-LOGS'
ORDER BY COLUMN_NAME;

AUTH_METHOD           TEXT
AVAILABLE_GB          NUMBER
AVAILABLE_MB          NUMBER
CONTENT_TYPE          TEXT
CURRENCY              TEXT
DISK_USAGE_PERCENT    NUMBER
FILE_SIZE_BYTES       NUMBER
MEMORY_PERCENT        NUMBER
METRICS_COUNT         NUMBER
PAYMENT_METHOD        TEXT
PRODUCT_ID            TEXT
PROVIDER              TEXT
QUERY_PARAMS          VARIANT
RATING                NUMBER
RECIPIENT             TEXT
REGION                TEXT
RESULT_COUNT          NUMBER
RETRY_COUNT           NUMBER
SESSION_DURATION      NUMBER
SMTP_CODE             NUMBER
STATUS                TEXT
TEST                  BOOLEAN
TIME_WINDOW           TEXT
TRACE_ID              TEXT
VALIDATION_ERRORS     VARIANT
VERSION               TEXT

🎉 WOW! These columns were added automatically without any DDL changes, pipeline reconfiguration, or downtime! This is automatic schema evolution in action!

Query Evolved Schema Fields

SELECT 
  TIMESTAMP,
  SERVICE,
  MESSAGE,
  REGION,
  TRACE_ID
FROM "APPLICATION-LOGS"
WHERE REGION IS NOT NULL
ORDER BY TIMESTAMP DESC
LIMIT 10;

SELECT 
  TIMESTAMP,
  SERVICE,
  USER_ID,
  AUTH_METHOD,
  PROVIDER,
  REGION
FROM "APPLICATION-LOGS"
WHERE AUTH_METHOD IS NOT NULL
ORDER BY TIMESTAMP DESC;

SELECT 
  TIMESTAMP,
  SERVICE,
  AMOUNT,
  CURRENCY,
  PAYMENT_METHOD,
  USER_ID,
  REGION
FROM "APPLICATION-LOGS"
WHERE CURRENCY IS NOT NULL
ORDER BY TIMESTAMP DESC;

SELECT 
  TIMESTAMP,
  SERVICE,
  MESSAGE,
  FILE_SIZE_BYTES,
  CONTENT_TYPE,
  REGION
FROM "APPLICATION-LOGS"
WHERE FILE_SIZE_BYTES IS NOT NULL
ORDER BY TIMESTAMP DESC;

SELECT 
  TIMESTAMP,
  SERVICE,
  MESSAGE,
  MEMORY_PERCENT,
  AVAILABLE_MB,
  DISK_USAGE_PERCENT,
  AVAILABLE_GB,
  REGION
FROM "APPLICATION-LOGS"
WHERE MEMORY_PERCENT IS NOT NULL 
   OR DISK_USAGE_PERCENT IS NOT NULL
ORDER BY TIMESTAMP DESC;

Note: System metrics only appear in WARN level logs with messages like "Memory usage above 80%" or "Disk space running low". If this query returns no results, produce more evolved logs to generate system metrics:

# Produce additional evolved logs
python $QUICK_START_REPO/sample-data/generate_logs.py --count 100 --evolved

Full Schema Evolution Queries: Run sql/2b.verify_schema_evolution.sql for comprehensive queries exploring all evolved fields including API parameters, error retries, email delivery, analytics metrics, and more!

Understanding What Just Happened

Real-World Benefits: This schema evolution capability is ideal for:

• Microservices with evolving log formats
• A/B testing with different instrumentation
• Gradual feature rollouts
• Multi-tenant systems with varying schemas
• DevOps teams that don't want to manage schema migrations

Troubleshooting

Tip: See the Monitoring the Pipeline chapter for detailed monitoring guidance.

Log Analytics

Basic Log Analytics

Count by Log Level

-- Distribution of log levels
SELECT 
  LEVEL as LOG_LEVEL,
  COUNT(*) as EVENT_COUNT,
  ROUND(COUNT(*) * 100.0 / SUM(COUNT(*)) OVER(), 2) as PERCENTAGE
FROM "APPLICATION-LOGS"
GROUP BY LOG_LEVEL
ORDER BY EVENT_COUNT DESC;

LOG_LEVEL | EVENT_COUNT | PERCENTAGE
----------|-------------|------------
INFO      | 350         | 70.00
WARN      | 100         | 20.00
ERROR     | 50          | 10.00

Note: The actual counts and percentages will vary depending on how many log events you've produced and the random distribution of log levels generated by the script.

Top Error Messages

-- Most common error messages
SELECT 
  MESSAGE as ERROR_MESSAGE,
  SERVICE,
  COUNT(*) as ERROR_COUNT
FROM "APPLICATION-LOGS"
WHERE LEVEL = 'ERROR'
GROUP BY ERROR_MESSAGE, SERVICE
ORDER BY ERROR_COUNT DESC
LIMIT 10;

Service Health Overview

-- Event count and error rate by service
SELECT 
  SERVICE as SERVICE_NAME,
  COUNT(*) as TOTAL_EVENTS,
  SUM(CASE WHEN LEVEL = 'ERROR' THEN 1 ELSE 0 END) as ERROR_COUNT,
  SUM(CASE WHEN LEVEL = 'WARN' THEN 1 ELSE 0 END) as WARN_COUNT,
  ROUND(ERROR_COUNT * 100.0 / NULLIF(TOTAL_EVENTS, 0), 2) as ERROR_RATE_PCT
FROM "APPLICATION-LOGS"
GROUP BY SERVICE_NAME
ORDER BY ERROR_RATE_PCT DESC;

Time-Series Analysis

Events Per Minute

-- Event volume over time (per minute)
SELECT 
  DATE_TRUNC('minute', TO_TIMESTAMP(TIMESTAMP)) as TIME_BUCKET,
  COUNT(*) as EVENTS_PER_MINUTE,
  SUM(CASE WHEN LEVEL = 'ERROR' THEN 1 ELSE 0 END) as ERRORS_PER_MINUTE
FROM "APPLICATION-LOGS"
WHERE TO_TIMESTAMP(TIMESTAMP) >= DATEADD('hour', -1, CURRENT_TIMESTAMP())
GROUP BY TIME_BUCKET
ORDER BY TIME_BUCKET DESC;

Peak Traffic Hours

-- Hourly event distribution
SELECT 
  HOUR(TO_TIMESTAMP(TIMESTAMP)) as HOUR_OF_DAY,
  COUNT(*) as EVENT_COUNT,
  AVG(DURATION_MS) as AVG_DURATION_MS
FROM "APPLICATION-LOGS"
WHERE TO_TIMESTAMP(TIMESTAMP) >= DATEADD('day', -1, CURRENT_TIMESTAMP())
  AND DURATION_MS IS NOT NULL
GROUP BY HOUR_OF_DAY
ORDER BY HOUR_OF_DAY;

Note: The TIMESTAMP column is stored as TEXT. We use TO_TIMESTAMP() to convert it to TIMESTAMP type for time-based analysis. If your quickstart was just completed, you may see limited data in the last hour/day time windows.

Performance Analytics

Slowest Requests

-- Slowest requests
SELECT
  SERVICE,
  REQUEST_ID,
  DURATION_MS,
  MESSAGE,
  TIMESTAMP AS REQUEST_TIME,
  LEVEL,
  STATUS_CODE
FROM
  QUICKSTART_KAFKA_CONNECTOR_DB.PUBLIC."APPLICATION-LOGS"
WHERE
  NOT DURATION_MS IS NULL
  AND DURATION_MS > 0
ORDER BY
  DURATION_MS DESC
LIMIT 20;

Performance by Service

-- Average performance metrics by service
SELECT
  SERVICE,
  COUNT(*) as TOTAL_REQUESTS,
  ROUND(AVG(DURATION_MS), 2) as AVG_DURATION_MS,
  MIN(DURATION_MS) as MIN_DURATION_MS,
  MAX(DURATION_MS) as MAX_DURATION_MS,
  SUM(
    CASE
      WHEN STATUS_CODE >= 400 THEN 1
      ELSE 0
    END
  ) as ERROR_COUNT,
  ROUND(
    AVG(
      CASE
        WHEN STATUS_CODE >= 400 THEN 1
        ELSE 0
      END
    ) * 100,
    2
  ) as ERROR_RATE_PCT
FROM
  QUICKSTART_KAFKA_CONNECTOR_DB.PUBLIC."APPLICATION-LOGS"
WHERE
  DURATION_MS IS NOT NULL
  AND DURATION_MS > 0
GROUP BY
  SERVICE
ORDER BY
  AVG_DURATION_MS DESC;

Advanced Analytics

Error Correlation

-- Find correlated errors (errors occurring together)
WITH ERROR_WINDOWS AS (
  SELECT
    SERVICE,
    CAST(TIMESTAMP AS TIMESTAMP) AS ERROR_TIME,
    REQUEST_ID,
    STATUS_CODE,
    DURATION_MS,
    DATE_TRUNC ('MINUTE', CAST(TIMESTAMP AS TIMESTAMP)) AS TIME_WINDOW
    /* Create 5-minute time windows */
  FROM
    QUICKSTART_KAFKA_CONNECTOR_DB.PUBLIC."APPLICATION-LOGS"
  WHERE
    LEVEL ILIKE '%ERROR%'
),
SERVICE_PAIRS AS (
  SELECT
    a.SERVICE AS SERVICE_A,
    b.SERVICE AS SERVICE_B,
    a.TIME_WINDOW,
    COUNT(DISTINCT a.REQUEST_ID) AS ERRORS_SERVICE_A,
    COUNT(DISTINCT b.REQUEST_ID) AS ERRORS_SERVICE_B,
    AVG(a.DURATION_MS) AS AVG_DURATION_A,
    AVG(b.DURATION_MS) AS AVG_DURATION_B
  FROM
    ERROR_WINDOWS AS a
    JOIN ERROR_WINDOWS AS b ON a.TIME_WINDOW = b.TIME_WINDOW
    AND a.SERVICE < b.SERVICE
    /* Avoid duplicate pairs */
  GROUP BY
    a.SERVICE,
    b.SERVICE,
    a.TIME_WINDOW
  HAVING
    ERRORS_SERVICE_A > 0
    AND ERRORS_SERVICE_B > 0
)
SELECT
  SERVICE_A,
  SERVICE_B,
  COUNT(*) AS CONCURRENT_ERROR_WINDOWS,
  SUM(ERRORS_SERVICE_A) AS TOTAL_ERRORS_A,
  SUM(ERRORS_SERVICE_B) AS TOTAL_ERRORS_B,
  ROUND(AVG(AVG_DURATION_A), 2) AS AVG_DURATION_A,
  ROUND(AVG(AVG_DURATION_B), 2) AS AVG_DURATION_B
FROM
  SERVICE_PAIRS
GROUP BY
  SERVICE_A,
  SERVICE_B
HAVING
  CONCURRENT_ERROR_WINDOWS > 1
ORDER BY
  CONCURRENT_ERROR_WINDOWS DESC,
  TOTAL_ERRORS_A + TOTAL_ERRORS_B DESC;

Why This Matters: Traditional log platforms charge premium prices for long retention and complex queries. With Snowflake, you can keep years of logs at low storage costs and run sophisticated SQL analytics without specialized query languages or expensive log indexing.

Semantic Search

Create Cortex Search Service

USE ROLE QUICKSTART_ROLE;
USE DATABASE QUICKSTART_KAFKA_CONNECTOR_DB;
USE SCHEMA PUBLIC;

-- Create Cortex Search service on log messages
-- Includes both base and evolved schema attributes for comprehensive filtering
CREATE OR REPLACE CORTEX SEARCH SERVICE application_logs_search
  ON MESSAGE
  ATTRIBUTES LEVEL, SERVICE, ERROR, STATUS_CODE, DURATION_MS, MEMORY_PERCENT, DISK_USAGE_PERCENT, REGION
  WAREHOUSE = QUICKSTART_KAFKA_CONNECTOR_WH
  TARGET_LAG = '1 minute'
  AS (
    SELECT 
      MESSAGE,
      LEVEL,
      SERVICE,
      ERROR,
      STATUS_CODE,
      DURATION_MS::NUMBER as DURATION_MS,
      TIMESTAMP,
      REQUEST_ID,
      HOST,
      USER_ID,
      MEMORY_PERCENT::NUMBER as MEMORY_PERCENT,
      AVAILABLE_MB::NUMBER as AVAILABLE_MB,
      DISK_USAGE_PERCENT::NUMBER as DISK_USAGE_PERCENT,
      AVAILABLE_GB::NUMBER as AVAILABLE_GB,
      REGION
    FROM "APPLICATION-LOGS"
  );

What This Does:

• ON MESSAGE: Indexes the MESSAGE field for semantic search
• ATTRIBUTES: LEVEL, SERVICE, ERROR, STATUS_CODE, DURATION_MS, MEMORY_PERCENT, DISK_USAGE_PERCENT, REGION become filterable dimensions
• TARGET_LAG: Search index updates within 1 minute of new data
• Semantic Understanding: Understands concepts, not just exact text matches
• Evolved Schema Support: Includes both base and evolved schema columns for comprehensive filtering
• Type Casting: Numeric columns (DURATION_MS, MEMORY_PERCENT, AVAILABLE_MB, DISK_USAGE_PERCENT, AVAILABLE_GB) are cast to NUMBER for proper filter operations
• Performance Filtering: DURATION_MS is indexed as an attribute, enabling fast filtering for performance-related searches (e.g., slow requests, timeout issues)

Query Using Snowflake Intelligence

Query Programmatically

Note: SEARCH_PREVIEW is designed for testing and validation. For production applications requiring low latency, use the Python API or the service's native REST endpoint.

-- Semantic search for authentication errors
-- Returns MESSAGE, LEVEL, SERVICE, TIMESTAMP, and ERROR columns
SELECT
  SNOWFLAKE.CORTEX.SEARCH_PREVIEW(
    'QUICKSTART_KAFKA_CONNECTOR_DB.PUBLIC.application_logs_search',
    '{
      "query": "authentication failed",
      "columns": ["MESSAGE", "LEVEL", "SERVICE", "TIMESTAMP", "ERROR"],
      "filter": {"@eq": {"LEVEL": "ERROR"}},
      "limit": 10
    }'
  ) AS search_results;

-- Search for payment issues with base schema columns
SELECT
  SNOWFLAKE.CORTEX.SEARCH_PREVIEW(
    'QUICKSTART_KAFKA_CONNECTOR_DB.PUBLIC.application_logs_search',
    '{
      "query": "payment declined timeout",
      "columns": ["MESSAGE", "SERVICE", "STATUS_CODE", "TIMESTAMP", "REQUEST_ID"],
      "filter": {"@eq": {"SERVICE": "payment-service"}},
      "limit": 10
    }'
  ) AS search_results;

-- Find database connection problems
SELECT
  SNOWFLAKE.CORTEX.SEARCH_PREVIEW(
    'QUICKSTART_KAFKA_CONNECTOR_DB.PUBLIC.application_logs_search',
    '{
      "query": "database connection timeout",
      "columns": ["MESSAGE", "LEVEL", "SERVICE", "HOST", "TIMESTAMP"],
      "filter": {"@eq": {"LEVEL": "ERROR"}},
      "limit": 10
    }'
  ) AS search_results;

Search with Multiple Filters

-- Find errors in payment OR auth services
SELECT
  SNOWFLAKE.CORTEX.SEARCH_PREVIEW(
    'QUICKSTART_KAFKA_CONNECTOR_DB.PUBLIC.application_logs_search',
    '{
      "query": "failed transaction",
      "columns": ["MESSAGE", "LEVEL", "SERVICE", "ERROR", "STATUS_CODE", "TIMESTAMP"],
      "filter": {
        "@and": [
          {"@eq": {"LEVEL": "ERROR"}},
          {"@or": [
            {"@eq": {"SERVICE": "payment-service"}},
            {"@eq": {"SERVICE": "auth-service"}}
          ]}
        ]
      },
      "limit": 20
    }'
  ) AS search_results;

-- Search for system warnings with high memory usage (evolved schema)
SELECT
  SNOWFLAKE.CORTEX.SEARCH_PREVIEW(
    'QUICKSTART_KAFKA_CONNECTOR_DB.PUBLIC.application_logs_search',
    '{
      "query": "memory usage warning system resource",
      "columns": ["MESSAGE", "LEVEL", "SERVICE", "MEMORY_PERCENT", "AVAILABLE_MB", "REGION", "TIMESTAMP"],
      "filter": {
        "@and": [
          {"@eq": {"LEVEL": "WARN"}},
          {"@gte": {"MEMORY_PERCENT": 80}}
        ]
      },
      "limit": 15
    }'
  ) AS search_results;

-- Search for slow requests with performance issues
SELECT
  SNOWFLAKE.CORTEX.SEARCH_PREVIEW(
    'QUICKSTART_KAFKA_CONNECTOR_DB.PUBLIC.application_logs_search',
    '{
      "query": "slow request timeout performance",
      "columns": ["MESSAGE", "SERVICE", "DURATION_MS", "STATUS_CODE", "TIMESTAMP", "REQUEST_ID"],
      "filter": {
        "@gte": {"DURATION_MS": 1000}
      },
      "limit": 20
    }'
  ) AS search_results;

Benefits of Cortex Search:

• Natural Language: Query logs in plain English, not complex regex
• Semantic Understanding: Finds conceptually similar messages, not just keyword matches
• Fast: Optimized vector search returns results in milliseconds
• Always Fresh: Automatically indexes new logs as they arrive
• Snowflake Intelligence Integration: Works seamlessly for conversational queries

Monitoring the Pipeline

Monitor Real-Time in Openflow Canvas

Statistics Reset: The statistics displayed on the canvas reset every 5 minutes. If you see zeros, it means no data has flowed in the current 5-minute window. This is normal between data generation runs or during idle periods.

Drill Down for Detailed Statistics

Healthy Pipeline Indicators:

• Minimal queue buildup (< 100 FlowFiles queued)
• Steady data throughput matching your production rate
• All processors in running state (green)
• No errors in processor logs

If you see large queues (> 1000), consider scaling up your runtime or adjusting processor concurrency settings.

Troubleshooting

Set Up Alerts (Optional)

USE ROLE QUICKSTART_ROLE;
USE DATABASE QUICKSTART_KAFKA_CONNECTOR_DB;

-- Alert if error rate exceeds threshold
CREATE OR REPLACE ALERT PUBLIC.HIGH_ERROR_RATE_ALERT
  WAREHOUSE = QUICKSTART_KAFKA_CONNECTOR_WH
  SCHEDULE = '5 MINUTE'
  IF (
    SELECT 
      SUM(CASE WHEN LEVEL = 'ERROR' THEN 1 ELSE 0 END) * 100.0 / NULLIF(COUNT(*), 0)
    FROM "APPLICATION-LOGS"
    WHERE TO_TIMESTAMP(TIMESTAMP) >= DATEADD('minute', -5, CURRENT_TIMESTAMP())
  ) > 15  -- Alert if > 15% error rate
  THEN CALL SYSTEM$SEND_EMAIL(
    'DataOps Team',
    '[email protected]',
    'Kafka Pipeline Alert: Error rate exceeds 15% in last 5 minutes'
  );

-- Alert on critical service errors
CREATE OR REPLACE ALERT PUBLIC.CRITICAL_SERVICE_ALERT
  WAREHOUSE = QUICKSTART_KAFKA_CONNECTOR_WH
  SCHEDULE = '5 MINUTE'
  IF (
    SELECT COUNT(*)
    FROM "APPLICATION-LOGS"
    WHERE LEVEL = 'ERROR'
      AND SERVICE IN ('payment-service', 'auth-service')
      AND TO_TIMESTAMP(TIMESTAMP) >= DATEADD('minute', -5, CURRENT_TIMESTAMP())
  ) > 10
  THEN CALL SYSTEM$SEND_EMAIL(
    'DataOps Team',
    '[email protected]',
    'Kafka Pipeline Alert: Critical service experiencing multiple errors'
  );

Customize Alerts: Adjust thresholds, services, and notification methods based on your operational needs. You can also integrate with Snowflake notifications for Slack, PagerDuty, or webhook integrations.

Cleanup

Stop Openflow Connector

Clean Up Snowflake Objects (Optional)

-- Use ACCOUNTADMIN to drop objects
USE ROLE ACCOUNTADMIN;

-- Drop Cortex Search service (if created)
DROP CORTEX SEARCH SERVICE IF EXISTS QUICKSTART_KAFKA_CONNECTOR_DB.PUBLIC.application_logs_search;

-- Drop alerts (if created)
DROP ALERT IF EXISTS QUICKSTART_KAFKA_CONNECTOR_DB.PUBLIC.HIGH_ERROR_RATE_ALERT;
DROP ALERT IF EXISTS QUICKSTART_KAFKA_CONNECTOR_DB.PUBLIC.CRITICAL_SERVICE_ALERT;

-- Drop database (this removes all tables and data)
DROP DATABASE IF EXISTS QUICKSTART_KAFKA_CONNECTOR_DB;

-- Drop warehouse
DROP WAREHOUSE IF EXISTS QUICKSTART_KAFKA_CONNECTOR_WH;

-- Drop external access integration
DROP EXTERNAL ACCESS INTEGRATION IF EXISTS quickstart_kafka_connector_access;

-- Note: We don't drop QUICKSTART_ROLE as it may be used by other quickstarts

Warning: This will permanently delete all log data and configuration. Make sure you've exported anything you need before dropping objects.

Clean Up Openflow Runtime (Optional)

Keep Runtime for Other Connectors: If you plan to use other Openflow connectors (PostgreSQL CDC, Google Drive, etc.), keep your runtime and deployment—they can be reused across multiple connectors!

Clean Up Kafka Resources (Optional)

# Delete the topic using rpk
rpk topic delete application-logs

Managed Kafka Services: For Confluent Cloud and other managed Kafka platforms, you can also delete topics through the service console or web UI.

Conclusion And Resources

What You Learned

Key Takeaways

Next Steps

Resources

Feedback