Sub-Second Contextualization At Scale: Turning Raw SCADA Data into Analyst-Ready Rows with Flink and Snowflake

How to stream industrial sensor data from a historian into Snowflake with full business context — enriched in-flight at hundreds of records per second

Key Contributor:

Bosco Albuquerque, Senior Partner Solutions Architect, AWS

Overview

A working pipeline that takes raw industrial sensor readings (flow rates, pressures, temperatures) and automatically adds business context — which company, facility, region, and contract each sensor belongs to — before the data lands in the analytics warehouse.

Why it matters: Without this, analysts receive millions of data points labeled with cryptic IDs like FLOW_0042. To answer basic business questions ("What's our total throughput in the Permian Basin this hour?"), they'd need to manually join reference data at query time — slow, error-prone, and expensive at scale.

What was demonstrated: A fully automated pipeline streaming up to 2,500 sensor readings per second from an industrial historian (DataPARC), through Amazon Kinesis and Apache Flink, into Snowflake. Flink enriches each record in-flight with producer, facility, region, and contract information using in-memory lookups. The pipeline processing — from data generation through enrichment to Snowpipe handoff — completes in under 500 milliseconds. Data becomes queryable in Snowflake within seconds, governed by Snowpipe Streaming’s commit cycle.

The business benefit: Data teams query fully contextualized operational data within seconds of generation — no batch jobs, no complex JOINs, no missing context. Tested at 100, 1,000, and 2,500 records per second with consistent sub-second pipeline performance at every volume.

What’s a Historian?

Industrial historians like DataPARC collect sensor data at scale — flow rates, pressures, temperatures — tagged with IDs like FLOW_0042 or TEMP_0117. Here's what a raw message looks like:

{
  "gen_timestamp": "2026-04-06T16:30:00.123Z",
  "tag_id": "FLOW_0042",
  "value": 1250.5,
  "unit": "BBL/DAY",
  "quality": "GOOD"
}

That’s it. No producer name. No facility. No region. No contract.

For an operations engineer, FLOW_0042 means something. For a data analyst in Snowflake trying to answer "what's the total throughput for Permian Basin contracts this hour?" — it means nothing.

The business context (which producer, facility, region, and contract each tag belongs to) lives in Snowflake as reference data. The challenge: how do you join that context onto every sensor reading before it lands in Snowflake, continuously, at hundreds of records per second — without adding a separate ETL step or a post-load JOIN?

The Architecture

Raw SCADA telemetry flows from the DataPARC historian into Amazon Kinesis, where Amazon Managed Service for Apache Flink picks it up, joins each record against dimension tables cached from Snowflake (via JDBC), and writes the fully contextualized rows back into Snowflake through the Snowpipe Streaming High-Performance SDK — all within seconds of the original sensor reading.

IoT pipeline using In-Flight Enrichment with Snowflake and Amazon Managed Service for Apache Flink

Below is a detailed breakdown of what is really happening under the hood in the demo we are deploying in this blog.

DataPARC Historian (OPC-UA source)
        │  5 fields: tag_id, value, unit, quality, timestamp
        │  ~500 records/sec via PutRecords (boto3)
        ▼
Amazon Kinesis Data Stream
        │  ON_DEMAND, partition key = tag_id
        ▼
Amazon Managed Service for Apache Flink (4 KPU, Flink 1.18.1)
  ├── ScadaDeserializer    → JSON to typed POJO + timestamp stamping
  ├── DimensionCache       → JDBC bulk load from Snowflake, 5-min refresh
  ├── EnrichmentFunction   → 2-phase context resolution (tag → IDs → names)
  └── SnowpipeSink         → High-Performance Snowpipe Streaming SDK, appendRow()
        │  16 fully enriched fields per record
        ▼
Snowflake (DATAPARC_DB.RAW)
  ├── DATAPARC_SCADA_VOLUMETRIC_V2   ← sink (enriched rows land here)
  │     Clustered by (INGESTION_TIME, TAG_ID)
  └── Enrichment Tables (read by Flink via JDBC):
        ├── TAG_CONTEXT         (tag_id → facility_id, producer_id, region_id, contract_id)
        ├── DIM_FACILITIES      (facility_id → facility_name, location)
        ├── DIM_PRODUCERS       (producer_id → producer_name, asset_class)
        ├── DIM_REGIONS         (region_id → region_name, country)
        └── DIM_CONTRACTS       (contract_id → contract_ref, terms)

Flink’s role in this pipeline is not just routing — it contextualizes every raw reading with business metadata before it reaches Snowflake, eliminating the need for post-load enrichment entirely. The key insight: Snowflake is both the source of enrichment metadata and the sink for enriched data. Flink reads the reference tables from Snowflake at startup and refreshes every 5 minutes. It writes fully enriched records back via the High-Performance Snowpipe Streaming SDK. Analysts query pre-joined rows — no runtime JOINs, no late-arriving NULLs.

The Enrichment Logic

Raw records carry only a tag_id. Flink performs two-phase enrichment entirely in memory:

Phase 1 — Tag Context Resolution: The DimensionCache holds a ConcurrentHashMap loaded via JDBC from TAG_CONTEXT (~1,000 rows). Each tag_id maps to four business IDs: facility_id, producer_id, region_id, contract_id.

Phase 2 — Dimension Name Lookup: Four small tables (10 rows each) map those IDs to human-readable names: producer name, asset name, region, contract terms.

Both phases are in-memory lookups — sub-millisecond per record. The entire reference dataset is ~1,040 rows across five tables; it fits in JVM heap with room to spare. The ScheduledExecutorService refreshes all maps every 5 minutes in the background without blocking record processing.

What lands in Snowflake for FLOW_0042:

{
  "tag_id": "FLOW_0042",
  "value": 1250.5,
  "quality": "GOOD",
  "producer_name": "Permian Basin Energy LLC",
  "asset_name": "Eagle Ford Central Processing",
  "region_name": "Permian Basin",
  "contract_terms": "Fixed-price gathering agreement",
  "gen_timestamp": "2026-04-06T16:30:00.123Z",
  "kinesis_approximate_arrival_timestamp": "2026-04-06T16:30:00.230Z",
  "flink_processing_timestamp": "2026-04-06T16:30:00.450Z",
  "ingestion_time": "2026-04-06T16:30:00.451Z",
  ...
}

All business context was added by Flink — not by DataPARC, and not by a post-load transformation.

Step-by-Step Deployment

Download and unzip the demo project:

curl -LO https://snowflake-corp-se-workshop.s3.us-west-1.amazonaws.com/dataparc-flink-demo/dataparc-flink-demo.zip
unzip dataparc-flink-demo.zip -d dataparc-flink-demo
cd dataparc-flink-demo
cp .env.example .env
# Edit .env with your Snowflake account, user, and AWS profile

The zip contains all Python scripts, the Java Flink project (pom.xml + 7 source files), and the .env.example template.

Prerequisites

• AWS account with permissions for Kinesis, Managed Flink, S3, IAM, Secrets Manager
• Snowflake account with SYSADMIN access
• Python 3.12+, Java 11+, Maven 3.9+
• RSA key pair for Snowflake key-pair authentication

Step 1: Generate RSA Key Pair and Register with Snowflake

# Generate private key (PKCS8, unencrypted)
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

In Snowflake:

ALTER USER <your_username>
  SET RSA_PUBLIC_KEY='<paste public key content without PEM headers>';

Step 2: Store Credentials in AWS Secrets Manager

aws secretsmanager create-secret \
  --name dataparc-flink-snowflake-credentials-v2 \
  --secret-string '{
    "SNOWFLAKE_ACCOUNT": "<account-identifier>",
    "SNOWFLAKE_USER": "<username>",
    "SNOWFLAKE_URL": "<account>.snowflakecomputing.com",
    "SNOWFLAKE_PRIVATE_KEY_CONTENT": "<base64 pkcs8 key without PEM headers>",
    "SNOWFLAKE_DATABASE": "DATAPARC_DB",
    "SNOWFLAKE_SCHEMA": "RAW",
    "SNOWFLAKE_WAREHOUSE": "<warehouse>"
  }'

The SNOWFLAKE_URL field is required by the High-Performance SDK — and must NOT include https://. Just <account>.snowflakecomputing.com.

Step 3: Set Up Snowflake Schema

python setup_snowflake.py --clean

Step 4: Create the Kinesis Stream

python setup_kinesis.py

Creates dataparc-scada-volumetric-stream-v2 in ON_DEMAND capacity mode. Partition key is tag_id — all readings from the same sensor go to the same shard, preserving order.

Step 5: Build the Flink Fat JAR

cd flink-processor
mvn clean package -DskipTests
# Output: target/dataparc-flink-1.0-SNAPSHOT.jar (~125 MB)

Step 6: Deploy to Amazon Managed Flink

python deploy_flink.py

This handles:

1. Creates S3 bucket and uploads the fat JAR
2. Creates IAM execution role with Kinesis, S3, Secrets Manager, and CloudWatch permissions
3. Creates the MSF application with Flink 1.18.1, 4 KPU, 60-second checkpointing

Step 7: Start the Flink Application

python deploy_flink.py --start

Wait 2–3 minutes for RUNNING status. Check CloudWatch Logs at /aws/kinesis-analytics/dataparc-scada-flink-processor-v2 if it stalls.

Step 8: Run the Simulator

python simulator.py --tags 100 --hz 5 --duration 60

Generates 500 records/sec (100 tags × 5 Hz) for 60 seconds, simulating a DataPARC historian publishing OPC-UA readings to Kinesis.

Step 9: Query Snowflake

At this point, the destination table has been enriched in-flight — each raw SCADA reading now carries its full dimensional context (producer, region, equipment class, etc.) without any post-load transformation. Query the table directly to confirm data is flowing and enrichment columns are populated:

-- Check record count
SELECT COUNT(*) FROM DATAPARC_DB.RAW.DATAPARC_SCADA_VOLUMETRIC_V2;

-- Verify enrichment landed correctly
SELECT TAG_ID, PRODUCER_NAME, REGION_NAME, VALUE, QUALITY
FROM DATAPARC_DB.RAW.DATAPARC_SCADA_VOLUMETRIC_V2
LIMIT 10;

-- Pipeline health view
SELECT * FROM DATAPARC_DB.RAW.VW_LATENCY_REALTIME_V2;

Cleanup

# Stop and delete Flink app (~$0.44/hr while running)
python deploy_flink.py --drop

# Delete Kinesis stream, IAM role, and Secrets Manager secret
python setup_kinesis.py --delete

# Drop Snowflake database and role
python setup_snowflake.py --drop

# Delete S3 bucket (JAR + enrichment file)
aws s3 rb s3://<your-jar-bucket> --force --region <your-aws-region>

After cleanup, all AWS and Snowflake resources are deleted. No recurring charges.

Conclusion

We built an end-to-end streaming pipeline that takes raw industrial sensor readings from a DataPARC historian, enriches them in-flight with business context using Apache Flink, and delivers fully contextualized records into Snowflake — all in under 500 milliseconds.

The key design decision: using Flink’s long-running JVM to maintain an in-memory cache of reference data (loaded from Snowflake via JDBC), so every record arrives in the warehouse pre-joined with producer, facility, region, and contract information. Analysts query ready-to-use data — no post-load transformations.