Overview

What You Will Learn

What You Will Build

What is Cortex Code?

Prerequisites

Prefer the fast path? If you have Cortex Code installed, skip ahead to Automate With Cortex Code and type CDR demo. It handles Kafka setup, Snowflake objects, config, build, and ML forecasting automatically.

Automate With Cortex Code

Install the Skill

git clone https://github.com/snowflakedb/snowpipe-streaming-sdk-examples.git
mkdir -p ~/.snowflake/cortex/skills
cp -r snowpipe-streaming-sdk-examples/custom-kafka-consumer/.cortex/skills/custom-kafka-consumer \
      ~/.snowflake/cortex/skills/

Available Triggers

Example Prompts

CDR demo

kafka to snowflake

set up the kafka consumer and stream some CDR data

run the streaming ingest kafka demo

What Gets Automated

Set Up Snowflake

Create Database Objects

USE ROLE ACCOUNTADMIN;

CREATE DATABASE IF NOT EXISTS CDR_STREAMING_DB;
CREATE SCHEMA IF NOT EXISTS CDR_STREAMING_DB.TELECOM;

CREATE TABLE IF NOT EXISTS CDR_STREAMING_DB.TELECOM.CALL_DETAIL_RECORDS (
    RECORD_ID         NUMBER,
    CALLER_NUMBER     VARCHAR,
    CALLEE_NUMBER     VARCHAR,
    CALL_TYPE         VARCHAR,   -- VOICE, SMS, DATA, MMS, VOICEMAIL
    NETWORK_TYPE      VARCHAR,   -- 4G, 5G, 3G, WIFI, ROAMING
    CALL_DISPOSITION  VARCHAR,   -- ANSWERED, NO_ANSWER, BUSY, FAILED, DROPPED
    DURATION_SECONDS  NUMBER,
    DATA_USAGE_MB     NUMBER,
    AREA_CODE         VARCHAR,
    CELL_TOWER        VARCHAR,   -- e.g. LAX-001, NYC-010
    PLAN              VARCHAR,   -- BASIC, STANDARD, PREMIUM, UNLIMITED, PREPAID
    EVENT_TIMESTAMP   TIMESTAMP_NTZ,
    INGEST_TIMESTAMP  TIMESTAMP_NTZ DEFAULT CURRENT_TIMESTAMP()
);

Configure RSA Key-Pair Authentication

# Generate private key (2048-bit, PKCS8 format)
openssl genrsa 2048 | openssl pkcs8 -topk8 -inform PEM -out rsa_key.p8 -nocrypt

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

# Print public key content (strip header/footer for Snowflake)
cat rsa_key.pub

ALTER USER <your_username> SET RSA_PUBLIC_KEY='<paste_public_key_content_here>';

-- Verify it was set
DESCRIBE USER <your_username>;

Security note: Never commit rsa_key.p8 or profile.json to version control. The repo's .gitignore already excludes profile.json.

Create profile.json

{
  "url": "https://<account_identifier>.snowflakecomputing.com",
  "account": "<account_identifier>",
  "user": "<your_username>",
  "role": "ACCOUNTADMIN",
  "warehouse": "COMPUTE_WH",
  "private_key": "<contents_of_rsa_key.p8_without_header_footer>"
}

Clone and Build

Clone the Repository

git clone https://github.com/snowflakedb/snowpipe-streaming-sdk-examples.git
cd snowpipe-streaming-sdk-examples/custom-kafka-consumer

Understand the Project Structure

custom-kafka-consumer/
├── consumer-config.properties   # Kafka + Snowflake target config
├── producer-config.properties   # Kafka broker + tuning config
├── profile.json                 # Snowflake credentials (gitignored)
├── pom.xml
└── src/main/java/com/snowflake/streaming/
    ├── consumer/
    │   ├── Main.java              # Entry point, launches N consumer threads
    │   ├── CustomKafkaConsumer.java  # Core consumer logic
    │   └── Config.java            # Loads consumer-config.properties
    └── producer/
        ├── FakeKafkaWriter.java   # Interactive CDR generator
        └── Config.java            # Loads producer-config.properties

<dependency>
    <groupId>com.snowflake</groupId>
    <artifactId>snowpipe-streaming</artifactId>
    <version>1.3.0</version>
</dependency>

Configure the Consumer

# Kafka
kafka.bootstrap.servers=localhost:9092
kafka.topic=cdr-topic
kafka.group.id=cdr-consumer-group
kafka.poll.duration.ms=1000

# Snowflake channel prefix
snowflake.channel.name=CDR_CHANNEL
snowflake.database=CDR_STREAMING_DB
snowflake.schema=TELECOM
snowflake.table=CALL_DETAIL_RECORDS

# Path to Snowflake profile JSON
snowflake.profile.path=profile.json

# Consumer tuning
max.rows.per.append=100
consumer.thread.count=3

Build

mvn clean compile

Start Kafka

brew install kafka
brew services start kafka

Set Up the Topic

kafka-topics.sh --create \
  --topic cdr-topic \
  --partitions 3 \
  --replication-factor 1 \
  --bootstrap-server localhost:9092

kafka-topics.sh --describe --topic cdr-topic --bootstrap-server localhost:9092

How the Consumer Works

1:1 Partition-to-Channel Mapping

consumer.subscribe(List.of(kafkaTopicName), new ConsumerRebalanceListener() {
    @Override
    public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
        for (TopicPartition tp : partitions) {
            String channelName = sfChannelPrefix + "_P" + tp.partition();
            OpenChannelResult result = sfClient.openChannel(channelName);
            SnowflakeStreamingIngestChannel channel = result.getChannel();
            partitionChannels.put(tp.partition(), channel);

            // Resume from where Snowflake last confirmed
            String lastToken = channel.getLatestCommittedOffsetToken();
            if (lastToken != null) {
                long offset = Long.parseLong(lastToken);
                consumer.seek(tp, offset + 1);
            }
        }
    }

    @Override
    public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
        for (TopicPartition tp : partitions) {
            SnowflakeStreamingIngestChannel channel = partitionChannels.remove(tp.partition());
            if (channel != null && !channel.isClosed()) {
                channel.close(true, Duration.ofSeconds(30));
            }
        }
    }
});

Smart Offset Commits

private void commitKafkaOffsetsAfterSnowflakeConfirm() {
    Map<TopicPartition, OffsetAndMetadata> toCommit = new HashMap<>();
    for (Map.Entry<Integer, SnowflakeStreamingIngestChannel> entry : partitionChannels.entrySet()) {
        int partition = entry.getKey();
        SnowflakeStreamingIngestChannel channel = entry.getValue();
        String token = channel.getLatestCommittedOffsetToken();
        if (token != null && !token.equals(lastCommittedTokens.get(partition))) {
            long offset = Long.parseLong(token);
            toCommit.put(new TopicPartition(kafkaTopicName, partition),
                         new OffsetAndMetadata(offset + 1));
            lastCommittedTokens.put(partition, token);
        }
    }
    if (!toCommit.isEmpty()) {
        consumer.commitSync(toCommit);
    }
}

Retry Logic

Run the Pipeline

Terminal 1 — Start the Consumer

cd snowpipe-streaming-sdk-examples/custom-kafka-consumer
mvn compile exec:java -Dexec.mainClass="com.snowflake.streaming.consumer.Main"

INFO  Main - Starting 3 consumer thread(s). Topic=cdr-topic, Group=cdr-consumer-group
INFO  CustomKafkaConsumer - Opened channel 'CDR_CHANNEL_P0' for partition 0. Last committed offset: null
INFO  CustomKafkaConsumer - Opened channel 'CDR_CHANNEL_P1' for partition 1. Last committed offset: null
INFO  CustomKafkaConsumer - Opened channel 'CDR_CHANNEL_P2' for partition 2. Last committed offset: null

Terminal 2 — Start the Producer

cd snowpipe-streaming-sdk-examples/custom-kafka-consumer
mvn compile exec:java -Dexec.mainClass="com.snowflake.streaming.producer.FakeKafkaWriter"

=== Fake Kafka Writer — Mobile CDR Generator ===
  Topic: cdr-topic

Commands:
  1 | single             Send one call record
  2 | burst [count]      Send a burst of CDRs (default 100)
  3 | stream [rps]       Continuous CDR stream (default 10/sec, Enter to stop)
  4 | malformed          Send malformed JSON messages
  5 | nulls              Send CDRs with null / missing fields
  6 | custom [json]      Send a custom JSON payload

> burst 500
Sent 500 CDRs in 312 ms (1603 rec/sec)

> stream 20
Streaming CDRs at ~20 rec/sec. Press Enter to stop...

Verify Data in Snowflake

SELECT COUNT(*) FROM CDR_STREAMING_DB.TELECOM.CALL_DETAIL_RECORDS;

SELECT CELL_TOWER, CALL_DISPOSITION, COUNT(*) AS RECORD_COUNT
FROM CDR_STREAMING_DB.TELECOM.CALL_DETAIL_RECORDS
GROUP BY 1, 2
ORDER BY 1, 2;

Test Fault Tolerance

INFO  CustomKafkaConsumer - Opened channel 'CDR_CHANNEL_P0'. Last committed offset: 499
INFO  CustomKafkaConsumer - Seeked partition 0 to offset 500

Train ML Forecast Models

Create Training Views

USE DATABASE CDR_STREAMING_DB;
USE SCHEMA TELECOM;

-- Daily call volume per tower
CREATE OR REPLACE VIEW V_DAILY_CALL_VOLUME AS
SELECT
    DATE_TRUNC('DAY', EVENT_TIMESTAMP)::TIMESTAMP_NTZ AS EVENT_DATE,
    CELL_TOWER,
    COUNT(*) AS CALL_COUNT
FROM CALL_DETAIL_RECORDS
WHERE EVENT_TIMESTAMP IS NOT NULL
GROUP BY 1, 2;

-- Daily call drop rate per tower
CREATE OR REPLACE VIEW V_DAILY_CALL_DROP_RATE AS
SELECT
    DATE_TRUNC('DAY', EVENT_TIMESTAMP)::TIMESTAMP_NTZ AS EVENT_DATE,
    CELL_TOWER,
    ROUND(
        SUM(CASE WHEN CALL_DISPOSITION = 'DROPPED' THEN 1 ELSE 0 END)::FLOAT
        / NULLIF(COUNT(*), 0),
        4
    ) AS DROP_RATE
FROM CALL_DETAIL_RECORDS
WHERE EVENT_TIMESTAMP IS NOT NULL
GROUP BY 1, 2;

-- Daily data usage per tower (sum of DATA call durations as proxy for MB)
CREATE OR REPLACE VIEW V_DAILY_DATA_USAGE AS
SELECT
    DATE_TRUNC('DAY', EVENT_TIMESTAMP)::TIMESTAMP_NTZ AS EVENT_DATE,
    CELL_TOWER,
    SUM(COALESCE(DATA_USAGE_MB, 0)) AS TOTAL_DATA_MB
FROM CALL_DETAIL_RECORDS
WHERE EVENT_TIMESTAMP IS NOT NULL
  AND CALL_TYPE = 'DATA'
GROUP BY 1, 2;

Train the Forecast Models

-- Model 1: Call volume forecast
CREATE OR REPLACE SNOWFLAKE.ML.FORECAST CALL_VOLUME_FORECAST(
    INPUT_DATA => SYSTEM$REFERENCE('VIEW', 'V_DAILY_CALL_VOLUME'),
    SERIES_COLNAME => 'CELL_TOWER',
    TIMESTAMP_COLNAME => 'EVENT_DATE',
    TARGET_COLNAME => 'CALL_COUNT'
);

-- Model 2: Call drop rate forecast
CREATE OR REPLACE SNOWFLAKE.ML.FORECAST CALL_DROP_RATE_FORECAST(
    INPUT_DATA => SYSTEM$REFERENCE('VIEW', 'V_DAILY_CALL_DROP_RATE'),
    SERIES_COLNAME => 'CELL_TOWER',
    TIMESTAMP_COLNAME => 'EVENT_DATE',
    TARGET_COLNAME => 'DROP_RATE'
);

-- Model 3: Data usage forecast
CREATE OR REPLACE SNOWFLAKE.ML.FORECAST DATA_USAGE_FORECAST(
    INPUT_DATA => SYSTEM$REFERENCE('VIEW', 'V_DAILY_DATA_USAGE'),
    SERIES_COLNAME => 'CELL_TOWER',
    TIMESTAMP_COLNAME => 'EVENT_DATE',
    TARGET_COLNAME => 'TOTAL_DATA_MB'
);

Generate Forecasts

-- Forecast call volume for next 7 days
CALL CALL_VOLUME_FORECAST!FORECAST(
    FORECASTING_PERIODS => 7,
    CONFIG_OBJECT => {'prediction_interval': 0.95}
);

-- Forecast call drop rates
CALL CALL_DROP_RATE_FORECAST!FORECAST(
    FORECASTING_PERIODS => 7,
    CONFIG_OBJECT => {'prediction_interval': 0.95}
);

-- Forecast data usage
CALL DATA_USAGE_FORECAST!FORECAST(
    FORECASTING_PERIODS => 7,
    CONFIG_OBJECT => {'prediction_interval': 0.95}
);

Save Forecast Results

CREATE OR REPLACE TABLE CALL_VOLUME_PREDICTIONS AS
    SELECT * FROM TABLE(
        CALL_VOLUME_FORECAST!FORECAST(FORECASTING_PERIODS => 7)
    );

CREATE OR REPLACE TABLE CALL_DROP_RATE_PREDICTIONS AS
    SELECT * FROM TABLE(
        CALL_DROP_RATE_FORECAST!FORECAST(FORECASTING_PERIODS => 7)
    );

CREATE OR REPLACE TABLE DATA_USAGE_PREDICTIONS AS
    SELECT * FROM TABLE(
        DATA_USAGE_FORECAST!FORECAST(FORECASTING_PERIODS => 7)
    );

Identify At-Risk Towers

SELECT
    SERIES AS CELL_TOWER,
    ROUND(AVG(FORECAST), 4) AS AVG_PREDICTED_DROP_RATE,
    ROUND(MAX(FORECAST), 4) AS PEAK_PREDICTED_DROP_RATE,
    COUNT(*) AS FORECAST_DAYS
FROM CALL_DROP_RATE_PREDICTIONS
GROUP BY 1
ORDER BY 2 DESC
LIMIT 10;

Query with Cortex Analyst

Create the Semantic View

CREATE OR REPLACE SEMANTIC VIEW CDR_STREAMING_DB.TELECOM.TOWER_ANALYTICS_SV
  TABLES (
    CDR_STREAMING_DB.TELECOM.CALL_DETAIL_RECORDS AS CDR
      PRIMARY KEY (RECORD_ID)
      WITH SYNONYMS ('call records', 'CDR', 'call data'),
    CDR_STREAMING_DB.TELECOM.CALL_DROP_RATE_PREDICTIONS AS DROP_FORECAST
      WITH SYNONYMS ('drop rate forecast', 'predicted drop rates'),
    CDR_STREAMING_DB.TELECOM.CALL_VOLUME_PREDICTIONS AS VOLUME_FORECAST
      WITH SYNONYMS ('call volume forecast', 'predicted call volumes'),
    CDR_STREAMING_DB.TELECOM.DATA_USAGE_PREDICTIONS AS USAGE_FORECAST
      WITH SYNONYMS ('data usage forecast', 'predicted data usage')
  )
  RELATIONSHIPS (
    DROP_FORECAST (SERIES) REFERENCES CDR (CELL_TOWER),
    VOLUME_FORECAST (SERIES) REFERENCES CDR (CELL_TOWER),
    USAGE_FORECAST (SERIES) REFERENCES CDR (CELL_TOWER)
  )
  FACTS (
    CDR.DURATION_SECONDS WITH SYNONYMS ('call duration', 'duration in seconds'),
    CDR.DATA_USAGE_MB WITH SYNONYMS ('data usage', 'megabytes used'),
    DROP_FORECAST.FORECAST AS PREDICTED_DROP_RATE
      WITH SYNONYMS ('predicted drop rate', 'forecasted call drops'),
    VOLUME_FORECAST.FORECAST AS PREDICTED_CALL_VOLUME
      WITH SYNONYMS ('predicted call volume', 'forecasted call count'),
    USAGE_FORECAST.FORECAST AS PREDICTED_DATA_USAGE_MB
      WITH SYNONYMS ('predicted data usage', 'forecasted bandwidth')
  )
  DIMENSIONS (
    CDR.CELL_TOWER WITH SYNONYMS ('tower', 'cell tower', 'tower ID'),
    CDR.CALL_TYPE WITH SYNONYMS ('type of call'),
    CDR.NETWORK_TYPE WITH SYNONYMS ('network', '4G', '5G'),
    CDR.CALL_DISPOSITION WITH SYNONYMS ('call result', 'dropped', 'answered'),
    CDR.AREA_CODE WITH SYNONYMS ('area', 'region'),
    CDR.PLAN WITH SYNONYMS ('subscriber plan', 'service plan'),
    CDR.EVENT_TIMESTAMP WITH SYNONYMS ('call time', 'when')
  )
  METRICS (
    METRIC TOTAL_CALLS AS COUNT(CDR.RECORD_ID)
      WITH SYNONYMS ('total calls', 'call count'),
    METRIC CALL_DROP_RATE AS
      SUM(CASE WHEN CDR.CALL_DISPOSITION = 'DROPPED' THEN 1 ELSE 0 END)::FLOAT
      / NULLIF(COUNT(CDR.RECORD_ID), 0)
      WITH SYNONYMS ('drop rate', 'percentage dropped'),
    METRIC AVG_CALL_DURATION AS AVG(CDR.DURATION_SECONDS)
      WITH SYNONYMS ('average duration', 'mean call length'),
    METRIC TOTAL_DATA_USAGE_MB AS SUM(CDR.DATA_USAGE_MB)
      WITH SYNONYMS ('total data usage', 'total MB')
  );

Use Cortex Analyst in Snowsight

What cell towers will have the highest call drop rate in the next 7 days?

Which towers are most at risk and need maintenance?

What is the forecasted data usage per tower this week?

Show me call volume by network type for the LAX towers.

Which plan type has the most dropped calls?

Use Cortex Analyst via API

curl -X POST \
  "https://<account_identifier>.snowflakecomputing.com/api/v2/cortex/analyst/message" \
  -H "Authorization: Bearer <jwt_token>" \
  -H "Content-Type: application/json" \
  -d '{
    "messages": [
      {
        "role": "user",
        "content": [
          {
            "type": "text",
            "text": "Which towers will have the highest call drop rate this week?"
          }
        ]
      }
    ],
    "semantic_view": "CDR_STREAMING_DB.TELECOM.TOWER_ANALYTICS_SV"
  }'

Conclusion And Resources

What You Learned

Related Resources