Real-Time Streaming from Amazon Kinesis to Snowflake with Openflow

Introduction

Modern data platforms demand real-time ingestion pipelines that are reliable, scalable, and low-maintenance. Amazon Kinesis is a popular choice for capturing streaming data — from IoT sensors to application events to clickstreams — but getting that data into Snowflake with sub-second latency traditionally required custom KCL consumers, complex infrastructure, and ongoing operational burden.

Snowflake Openflow eliminates that complexity. Built on Apache NiFi, Openflow provides a visual, connector-based approach to streaming ingestion with two deployment modes:

• SPCS (Snowflake-managed) — runs as a fully managed service on Snowflake Container Services. No servers to provision, no JVMs to tune, no infrastructure to patch. Snowflake manages the runtime, scaling, and restarts for you.
• BYOC (Bring Your Own Compute) — runs on your own infrastructure managed by Amazon EKS. You manage the runtime but retain full control over networking, compute sizing, and deployment topology.

In this guide, we use the SPCS deployment for its zero-ops simplicity. You’ll build a complete end-to-end pipeline using the Openflow Kinesis connector, which internally leverages:

• Amazon Kinesis — the source data stream
• KCL 3.x (Kinesis Client Library) — built into the connector for shard consumption, fan-out, and rebalancing
• Amazon DynamoDB — used by KCL to store shard lease checkpoints (ensures at-least-once delivery and consumer coordination)
• Amazon CloudWatch — used by KCL to publish consumer lag and throughput metrics

What you’ll achieve:

• A Kinesis stream producing real-time flight tracking data
• An Openflow connector consuming records via KCL 3.x
• Automatic schema evolution in Snowflake — new JSON fields become new columns with zero DDL
• DynamoDB-backed checkpointing for at-least-once delivery
• A visual NiFi canvas for monitoring and debugging

Press enter or click to view image in full size

Why Openflow for Kinesis Ingestion?

There are multiple ways to get data from Kinesis into Snowflake. Here’s how Openflow compares to code-based alternatives:

Openflow provides a no-code approach to streaming data from Kinesis into Snowflake, while delivering the same high-performance ingestion as code-based solutions. Under the hood, the Openflow Kinesis connector uses the Snowpipe Streaming SDK — the same battle-tested library that powers custom integrations — but wraps it in a visual, configuration-driven interface that eliminates the need for application code.

Key Benefits:

• Zero Code Required — Configure the entire pipeline through the NiFi canvas UI. No Java, Python, or SDK knowledge needed. Just point at your Kinesis stream, map fields to Snowflake columns, and start the connector.
• High-Performance SDK Under the Hood — The PutSnowpipeStreaming processor uses the official Snowpipe Streaming High Performance Architecture SDK (same as custom integrations), delivering the same sub-second latency and high throughput. You get SDK-level performance without writing SDK code.
• Fully Managed Infrastructure — Runs as a Snowflake SPCS service. No EC2 instances to provision, no JVMs to tune, no patching. Snowflake handles restarts, scaling, and availability.
• Built-in KCL Integration — The Kinesis consumer uses AWS KCL 3.x with automatic checkpointing to DynamoDB. Handles shard rebalancing, consumer coordination, and at-least-once delivery out of the box.
• Automatic Schema Evolution — New fields in your JSON data automatically create new columns in Snowflake. No manual DDL, no downtime for schema changes.
• Visual Monitoring & Debugging — The NiFi canvas provides real-time visibility into data flow, processor metrics, and error queues. Troubleshoot issues without parsing log files or writing custom CloudWatch dashboards.
• Shortest Time to Production — Deploy a working pipeline in under an hour. Compare this to days of development, testing, and deployment for a custom KCL consumer.

Prerequisites

Before building the demo pipeline, ensure you have the following:

Snowflake

• Snowflake account with Openflow deployed (Snowflake-managed SPCS runtime)
• ACCOUNTADMIN access (for creating roles, integrations, and grants)
• Openflow canvas URL — you’ll configure the connector via the web UI, supposed you already have a runtime created in a SPCS deployment for this lab.

Verify your Openflow deployment:

-- Data plane integration must exist and be enabled
SHOW OPENFLOW DATA PLANE INTEGRATIONS;

-- Runtime integration must exist with snowflakecomputing.app in the OAuth URL
-- (If only snowflake-customer.app URLs exist, those are BYOC — deploy a Snowflake-managed runtime first)
SHOW OPENFLOW RUNTIME INTEGRATIONS;

Note: Snowflake-managed Openflow runtimes are control-plane-managed and do NOT appear in SHOW SERVICES. The correct verification is that a runtime integration exists with snowflakecomputing.app in its oauth_redirect_uri.

AWS

• AWS account with an IAM user or role that has permissions for:

kinesis:* (create/describe/put/get records)
dynamodb:* (KCL checkpoint table)
cloudwatch:PutMetricData (KCL metrics)

• AWS CLI installed and configured with a named profile
• Python 3.8+ with boto3 and requests packages (for the test producer)

Step-by-Step Instructions

Step 1: Create a Kinesis Data Stream

Create a single-shard stream. One shard provides up to 1 MB/s ingest — more than enough for this demo.

aws kinesis create-stream \
  --stream-name <YOUR_KINESIS_STREAM_NAME> \
  --shard-count 1 \
  --region <YOUR_AWS_REGION> \
  --profile <YOUR_AWS_PROFILE>

# Wait for the stream to become ACTIVE (~30 seconds)
aws kinesis wait stream-exists \
  --stream-name <YOUR_KINESIS_STREAM_NAME> \
  --region <YOUR_AWS_REGION> \
  --profile <YOUR_AWS_PROFILE> && echo "Stream is ACTIVE"

Tip: For production workloads, consider switching to ON_DEMAND mode (auto-scales shards) via the AWS Console after creation.

Step 2: Start a Test Producer

We’ll use OpenSky flight tracking data to validate the pipeline. The producer script supports flags for controlling how many records to send and optionally adding extra fields (used later to test schema evolution).

Download the producer script, save it as opensky_producer.py and install the depencies:

pip install boto3 requests

Send 500 records to seed the pipeline:

python3 opensky_producer.py \
  --stream <YOUR_KINESIS_STREAM_NAME> \
  --profile <YOUR_AWS_PROFILE> \
  --region <YOUR_AWS_REGION> \
  --count 500

You should see output like:

Producing to: <YOUR_KINESIS_STREAM_NAME> | Region: us-west-2
Mode: send 500 records then exit
--------------------------------------------------
[14:23:45] Sent 500 records (total: 500)

Done. Sent 500 records.

The producer exits after sending the specified count. Continue to the next steps while the records flow through the pipeline.

Running in production: The local producer is for testing only. For continuous, scalable ingestion in production, deploy the producer to a managed AWS service like Lambda, ECS, EKS or EC2.

Step 3: Identify the Openflow Role

The Openflow runtime authenticates to Snowflake via OAuth using a base role — the role that owns the data plane integration. Discover it:

-- Find the data plane integration(run time) name that uses SPCS
  USE ROLE ACCOUNTADMIN;

  SHOW OPENFLOW RUNTIME INTEGRATIONS;

  SELECT
    "name",
    CASE
      WHEN "oauth_redirect_uri" LIKE '%snowflakecomputing.app%' THEN 'SPCS'
      WHEN "oauth_redirect_uri" LIKE '%snowflake-customer.app%' THEN 'BYOC'
      ELSE 'UNKNOWN'
    END AS deployment_type,
    "oauth_redirect_uri"
  FROM TABLE(RESULT_SCAN(LAST_QUERY_ID()))
  ORDER BY deployment_type, "name";

-- Find who owns it (look for the OWNERSHIP grant)
SHOW GRANTS ON INTEGRATION <YOUR_DATAPLANE_INTEGRATION>;
-- The role with OWNERSHIP is your Openflow base role, 
-- You need to use this role to access the Openflow UI later

Note this role name — you’ll grant it access to your destination database. You will also need it to access the Openflow UI in step 5.

Step 4: Create Snowflake Database and Table

Create a destination database and an empty table with schema evolution enabled. The connector will automatically add columns as it encounters new fields in your JSON data.

USE ROLE ACCOUNTADMIN;

-- Create destination database
CREATE DATABASE IF NOT EXISTS <YOUR_DATABASE>;

-- Create table with schema evolution enabled
-- The connector flattens JSON fields into individual columns automatically
CREATE TABLE IF NOT EXISTS <YOUR_DATABASE>.PUBLIC.<YOUR_TABLE> ()
  ENABLE_SCHEMA_EVOLUTION = TRUE;

-- Grant access to the Openflow role
GRANT USAGE ON DATABASE <YOUR_DATABASE> TO ROLE <YOUR_OPENFLOW_ROLE>;
GRANT USAGE ON SCHEMA <YOUR_DATABASE>.PUBLIC TO ROLE <YOUR_OPENFLOW_ROLE>;
GRANT USAGE ON WAREHOUSE <YOUR_WAREHOUSE> TO ROLE <YOUR_OPENFLOW_ROLE>;
GRANT CREATE TABLE ON SCHEMA <YOUR_DATABASE>.PUBLIC TO ROLE <YOUR_OPENFLOW_ROLE>;
GRANT ALL PRIVILEGES ON ALL TABLES IN SCHEMA <YOUR_DATABASE>.PUBLIC TO ROLE <YOUR_OPENFLOW_ROLE>;
GRANT ALL PRIVILEGES ON FUTURE TABLES IN SCHEMA <YOUR_DATABASE>.PUBLIC TO ROLE <YOUR_OPENFLOW_ROLE>;

Step 5: Create an External Access Integration

The Openflow connector (running on SPCS) needs network egress to reach Kinesis, DynamoDB (for KCL checkpoints), and CloudWatch (for metrics). Create a network rule and integration:

USE ROLE ACCOUNTADMIN;

CREATE OR REPLACE NETWORK RULE <YOUR_NETWORK_RULE>
  MODE = EGRESS
  TYPE = HOST_PORT
  VALUE_LIST = (
    -- Kinesis data plane
    'kinesis.<YOUR_AWS_REGION>.amazonaws.com:443',
    'kinesis.<YOUR_AWS_REGION>.api.aws:443',
    '*.control-kinesis.<YOUR_AWS_REGION>.amazonaws.com:443',
    '*.data-kinesis.<YOUR_AWS_REGION>.amazonaws.com:443',
    '*.control-kinesis.<YOUR_AWS_REGION>.api.aws:443',
    '*.data-kinesis.<YOUR_AWS_REGION>.api.aws:443',
    -- DynamoDB (KCL checkpoints — REQUIRED)
    'dynamodb.<YOUR_AWS_REGION>.amazonaws.com:443',
    -- CloudWatch (KCL metrics)
    'monitoring.<YOUR_AWS_REGION>.amazonaws.com:443',
    'monitoring.<YOUR_AWS_REGION>.api.aws:443'
  );

CREATE OR REPLACE EXTERNAL ACCESS INTEGRATION <YOUR_EAI_NAME>
  ALLOWED_NETWORK_RULES = ('<YOUR_NETWORK_RULE>')
  ENABLED = true;

GRANT USAGE ON INTEGRATION <YOUR_EAI_NAME> TO ROLE <YOUR_OPENFLOW_ROLE>;

Important: After creating the EAI, you MUST attach it to your Openflow runtime. The connector will fail with UnknownHostException errors if the EAI is not attached before starting.

1. Access the Openflow UI: login to Snowsight as a user with ACCOUNTADMIN role, then switch to the role noted in step 3 above. Navigate to Ingestion the left pane, select Openflow, then click the Launch Openflow button located at the top-right corner of the window. Press the Runtimes tab to show all available runtimes.
2. Find your SPCS runtime (deployment = snowspcs, state = Active)
3. Click the three-dot menu (⋮) on the right side of the runtime row
4. Select External access integrations
5. Add your EAI (<YOUR_EAI_NAME>) and save
6. Wait 30–60 seconds for the runtime to pick up the new network configuration

Press enter or click to view image in full size

Step 6: Deploy the Connector via Openflow UI

1. Click on the desired runtime to open its canvas.
2. Drag the Import from registry icon from the top toolbar onto the canvas — this opens the import dialog

Press enter or click to view image in full size

3. In the import dialog:

• Registry Client: ConnectorFlowRegistryClient
• Bucket: connectors
• Flow Name: kinesis-high-performance
• Version: select the latest available

4. Click Import — the connector process group appears on your canvas

5. Double-click the process group to go inside. You’ll see two processors connected:

• ConsumeKinesis — KCL consumer that reads from Kinesis
• PublishSnowpipeStreaming — writes records to Snowflake via Snowpipe Streaming

Step 7: Configure Connector Parameters

The connector uses a parameter context for all configuration. All values are set in one place.

1. Right-click on the canvas background → Parameters (or navigate via the hamburger menu ≡ → Parameter Contexts)
2. Click on the parameter context (named something like “Kinesis High Performance”) to edit it
3. Fill in the following values:

Press enter or click to view image in full size

Step 8: Enable Controller Services

After deploying and configuring, the connector’s controller services are disabled by default. You must enable them before starting.

1. Right-click canvas → Controller Services
2. You’ll see 4 services, all in “Disabled” state:

• AWSCredentialsProviderControllerService
• JsonTreeReader
• JsonRecordSetWriter
• StandardWebClientServiceProvider

3. Enable each service by clicking the ⚡ icon on the right side of each row (or use the three-dot menu → Enable)

4. All 4 should show Enabled state

Press enter or click to view image in full size

Step 9: Start the Connector

1. Navigate back to the canvas (click the process group breadcrumb at the top)
2. Confirm both processors show red stop squares (valid, stopped) — no yellow warning triangles
3. Right-click on the canvas → Start (or select all with Ctrl+A and click ▶ Play in the toolbar)

The connector is now running. KCL will initialize, acquire shard leases from DynamoDB, and begin consuming from Kinesis.

Note: Allow 2–5 minutes for KCL initialization on first run.

Step 10: Verify Data is Flowing

Check the canvas: After a few minutes, you should see:

• ConsumeKinesis: “Out” counter incrementing (bytes flowing out)
• Queue between processors: items briefly appearing and draining
• PublishSnowpipeStreaming: “In” counter incrementing

Query Snowflake for ingested data:

-- Check row count
SELECT COUNT(*) AS ROW_COUNT FROM <YOUR_DATABASE>.PUBLIC.<YOUR_TABLE>;

-- View the table schema (columns auto-created by schema evolution)
DESCRIBE TABLE <YOUR_DATABASE>.PUBLIC.<YOUR_TABLE>;

-- View recent records
SELECT * FROM <YOUR_DATABASE>.PUBLIC.<YOUR_TABLE>
  LIMIT 10;

Verify KCL checkpoints in DynamoDB:

aws dynamodb scan --table-name <YOUR_APPLICATION_NAME> \
  --region <YOUR_AWS_REGION> --profile <YOUR_AWS_PROFILE> \
  --query 'Items[*].{shard:leaseKey.S,checkpoint:checkpoint.S}'

If you see rows in Snowflake and active checkpoints in DynamoDB — your pipeline is working end-to-end.

Step 11: Schema Evolution

One of Openflow’s key features is automatic schema evolution — when new fields appear in your JSON data, Snowflake automatically adds columns without any DDL changes.

Run the same producer with the --extra-fields flag to send records that include a ground_speed field not yet in the table:

python3 opensky_producer.py \
  --stream <YOUR_KINESIS_STREAM_NAME> \
  --profile <YOUR_AWS_PROFILE> \
  --region <YOUR_AWS_REGION> \
  --extra-fields ground_speed \
  --count 5

Your should see output like this:

Producing to: <YOUR_KINESIS_STREAM_NAME> | Region: us-west-2
Extra fields: ground_speed
Mode: send 5 records then exit
--------------------------------------------------
[14:30:12] Sent 5 records (total: 5)

Done. Sent 5 records.

Wait 30 seconds, then verify the new column appeared:

-- Check for the new column
DESCRIBE TABLE <YOUR_DATABASE>.PUBLIC.<YOUR_TABLE>;
-- You should see GROUND_SPEED as a new VARCHAR column

-- Query the records with the new field
SELECT ID, ICAO, ORIG, DEST, GROUND_SPEED
FROM <YOUR_DATABASE>.PUBLIC.<YOUR_TABLE>
WHERE GROUND_SPEED IS NOT NULL;

The GROUND_SPEED column was automatically added by Snowpipe Streaming's schema evolution — no ALTER TABLE needed. Existing rows will show NULL for the new column (expected).

Summary

You’ve built a fully managed, real-time streaming pipeline that:

• Consumes data from Amazon Kinesis using KCL 3.x with DynamoDB-backed checkpointing
• Ingests records into Snowflake via Snowpipe Streaming (single-digit-second latency)
• Auto-evolves schema as new fields appear — no DDL required
• Runs on SPCS with zero infrastructure to manage — no EC2, no JVMs, no patching

Next Steps For Production Deployment

• Scale throughput: Switch AWS Kinesis Consumer Type to ENHANCED_FAN_OUT for dedicated 2 MB/s per shard per consumer
• Add monitoring: Set up CloudWatch alarms on MillisBehindLatest for consumer lag alerts
• Production hardening: Create a dedicated Openflow role with least-privilege grants
• Replace the test producer: Point your real data sources at the Kinesis stream

Cleanup

Stop and Delete the Openflow Connector:

1. Open the Openflow canvas UI
2. Right-click on the process group → Stop
3. Once stopped, right-click → Delete
4. Confirm deletion

Delete AWS Resources:

# Delete the KCL DynamoDB table
aws dynamodb delete-table --table-name <YOUR_APPLICATION_NAME> \
  --region <YOUR_AWS_REGION> --profile <YOUR_AWS_PROFILE>

# Delete the Kinesis stream
aws kinesis delete-stream --stream-name <YOUR_KINESIS_STREAM_NAME> \
  --enforce-consumer-deletion \
  --region <YOUR_AWS_REGION> --profile <YOUR_AWS_PROFILE>

Delete Snowflake Resources:

USE ROLE ACCOUNTADMIN;

-- Drop the table
DROP TABLE IF EXISTS <YOUR_DATABASE>.PUBLIC.<YOUR_TABLE>;

-- Drop the integration and network rule
DROP EXTERNAL ACCESS INTEGRATION IF EXISTS <YOUR_EAI_NAME>;
DROP NETWORK RULE IF EXISTS <YOUR_NETWORK_RULE>;

-- Optional: Drop the database if created for testing
-- DROP DATABASE IF EXISTS <YOUR_DATABASE>;

Detach EAI from Runtime:

In the Openflow Control Plane UI, detach the EAI from the runtime.

Verify Cleanup

# Confirm AWS resources are gone
aws kinesis list-streams --region <YOUR_AWS_REGION> --profile <YOUR_AWS_PROFILE>
aws dynamodb list-tables --region <YOUR_AWS_REGION> --profile <YOUR_AWS_PROFILE>

-- Confirm Snowflake resources are gone
SHOW TABLES IN <YOUR_DATABASE>.PUBLIC;
SHOW INTEGRATIONS LIKE '<YOUR_EAI_NAME>';