Overview

Prerequisites

What You'll Learn

What You'll Need

NOTE: Some AWS resources in this guide bill while they exist, including the Kinesis stream and the Firehose delivery streams. The Snowflake warehouse is X-Small and suspends after 60 seconds. Run the Clean Up step when you finish.

What You'll Build

How It Works

The Purchase Event

{
  "purchase_id":  "9f1c2a3e-6b7d-4c8e-9a10-2b3c4d5e6f70",
  "event_id":     "EVT001",
  "seat_id":      "A-101",
  "customer_id":  "CUST-8821",
  "price_paid":   150.00,
  "channel":      "web",
  "promo_code":   "EARLY20",
  "party_size":   4,
  "purchased_at": "2026-07-23T01:06:07.038Z"
}

The Venue

Prepare Snowflake

USE ROLE ACCOUNTADMIN;

CREATE DATABASE IF NOT EXISTS ticketing_db;
CREATE SCHEMA   IF NOT EXISTS ticketing_db.pipeline;

CREATE WAREHOUSE IF NOT EXISTS ticketing_wh
  WAREHOUSE_SIZE      = 'XSMALL'
  AUTO_SUSPEND        = 60
  AUTO_RESUME         = TRUE
  INITIALLY_SUSPENDED = TRUE;

USE WAREHOUSE ticketing_wh;
USE SCHEMA ticketing_db.pipeline;

Create the Bronze Tables

-- Batch path: loaded by Snowpipe from S3 files
CREATE OR REPLACE TABLE ticketing_db.pipeline.bronze_purchases (
  purchase_id    VARCHAR,
  event_id       VARCHAR,
  seat_id        VARCHAR,
  customer_id    VARCHAR,
  price_paid     NUMBER(10,2),
  channel        VARCHAR,
  promo_code     VARCHAR,
  party_size     NUMBER,
  purchased_at   TIMESTAMP_TZ,
  _loaded_at     TIMESTAMP_TZ DEFAULT CURRENT_TIMESTAMP(),
  _src_file      VARCHAR
);

-- Fast path: written directly by Firehose through Snowpipe Streaming
CREATE OR REPLACE TABLE ticketing_db.pipeline.bronze_purchases_rt (
  purchase_id    VARCHAR,
  event_id       VARCHAR,
  seat_id        VARCHAR,
  customer_id    VARCHAR,
  price_paid     NUMBER(10,2),
  channel        VARCHAR,
  promo_code     VARCHAR,
  party_size     NUMBER,
  purchased_at   TIMESTAMP_TZ
);

IMPORTANT: Do not add a _loaded_at column, or any column with a DEFAULT, to bronze_purchases_rt. Snowpipe Streaming, which the Firehose Snowflake destination uses, rejects a table that has a column with DEFAULT, AUTOINCREMENT, IDENTITY, GEO, or a collation. It does not skip the column. It fails the entire delivery with a Snowflake.InvalidColumns error.

Create AWS Streaming Layer

Create the Kinesis Data Stream

Create the S3 Bucket

Configure Key-Pair Auth

mkdir ticketing-pipeline
cd ticketing-pipeline
mkdir firehose-snowflake-auth
cd firehose-snowflake-auth

openssl genrsa 2048 | openssl pkcs8 -topk8 -inform PEM -out rsa_key.p8 -nocrypt
openssl rsa -in rsa_key.p8 -pubout -out rsa_key.pub

IMPORTANT: rsa_key.p8 is a private key. Never commit it to Git or share it. If you put this folder under version control, add rsa_key.p8 to .gitignore first.

Register the Public Key

ALTER USER <your_user>
  SET RSA_PUBLIC_KEY = '<paste base64 body here, no header, no footer, no line breaks>';

-- Verify: RSA_PUBLIC_KEY_FP and RSA_PUBLIC_KEY_LAST_SET_TIME should be populated
DESC USER <your_user>;

Find Your Account Identifier

Build the Fast Path

Create and Seed the Seats Table

CREATE OR REPLACE HYBRID TABLE ticketing_db.pipeline.seats (
  seat_id      VARCHAR PRIMARY KEY,
  event_id     VARCHAR,
  section      VARCHAR,
  row_number   NUMBER,
  seat_number  NUMBER,
  price_tier   VARCHAR,
  price        NUMBER(10,2),
  status       VARCHAR DEFAULT 'AVAILABLE',
  sold_at      TIMESTAMP_TZ
);

INSERT INTO ticketing_db.pipeline.seats
  (seat_id, event_id, section, row_number, seat_number, price_tier, price, status, sold_at)
SELECT
  section || '-' || (row_num::STRING) || LPAD(seat_num::STRING, 2, '0') AS seat_id,
  'EVT001', section, row_num, seat_num, price_tier, price, 'AVAILABLE', NULL
FROM (
  SELECT
    CHR(65 + FLOOR(idx / 20))         AS section,
    FLOOR(MOD(idx, 20) / 10) + 1      AS row_num,
    MOD(idx, 10) + 1                  AS seat_num,
    CASE
      WHEN FLOOR(idx / 20) <= 3  THEN 'FLOOR'   -- sections A-D
      WHEN FLOOR(idx / 20) <= 11 THEN 'LOWER'   -- sections E-L
      ELSE                            'UPPER'   -- sections M-P
    END AS price_tier,
    CASE
      WHEN FLOOR(idx / 20) <= 3  THEN 150.00
      WHEN FLOOR(idx / 20) <= 11 THEN  85.00
      ELSE                             45.00
    END AS price
  FROM (SELECT SEQ4() AS idx FROM TABLE(GENERATOR(ROWCOUNT => 320)))
);

SELECT COUNT(*) FROM ticketing_db.pipeline.seats;                          -- expect 320
SELECT section, price_tier, price, COUNT(*) FROM ticketing_db.pipeline.seats
  GROUP BY section, price_tier, price ORDER BY section;                    -- 20 per section
SELECT status, COUNT(*) FROM ticketing_db.pipeline.seats GROUP BY status;  -- 320 AVAILABLE

Create the Stream and the Task

CREATE OR REPLACE STREAM ticketing_db.pipeline.bronze_purchases_rt_stream
  ON TABLE ticketing_db.pipeline.bronze_purchases_rt;

CREATE OR REPLACE TASK ticketing_db.pipeline.seat_status_update_task
    WAREHOUSE = ticketing_wh
    USER_TASK_MINIMUM_TRIGGER_INTERVAL_IN_SECONDS = 10
    WHEN SYSTEM$STREAM_HAS_DATA('ticketing_db.pipeline.bronze_purchases_rt_stream')
AS
MERGE INTO ticketing_db.pipeline.seats AS tgt
USING (
    SELECT seat_id, purchased_at
    FROM (
        SELECT seat_id, purchased_at,
               ROW_NUMBER() OVER (PARTITION BY seat_id ORDER BY purchased_at DESC) AS rn
        FROM ticketing_db.pipeline.bronze_purchases_rt_stream
        WHERE METADATA$ACTION = 'INSERT'   -- ignore DELETE records created by TRUNCATE
    )
    WHERE rn = 1                            -- keep only the newest record per seat
) AS src
ON tgt.seat_id = src.seat_id
WHEN MATCHED THEN UPDATE SET
    tgt.status  = 'SOLD',
    tgt.sold_at = src.purchased_at;

-- Tasks are created suspended. Without this line the Task never runs.
ALTER TASK ticketing_db.pipeline.seat_status_update_task RESUME;

NOTE: The Task needs both safeguards in the MERGE. TRUNCATE TABLE writes DELETE records into the Stream, and without the METADATA$ACTION = 'INSERT' filter a DELETE and an INSERT for the same seat collide on the primary key. A backlog from failed runs can also hold the same seat_id more than once, which the ROW_NUMBER() filter resolves.

NOTE: 10 seconds is the minimum value for USER_TASK_MINIMUM_TRIGGER_INTERVAL_IN_SECONDS. The WHEN clause accepts only SYSTEM$STREAM_HAS_DATA and SYSTEM$GET_PREDECESSOR_RETURN_VALUE. Checking the Stream does not start the warehouse. The warehouse runs only when the Task actually fires.

Create the Fast-Path Firehose Stream

NOTE: S3 here is only a dead-letter location. The active data path is Firehose to Snowflake with no files in between. If deliveries fail, check the backup location. Each failed record includes the exact error message, for example Snowflake.InvalidColumns, and the original payload. Enter only the bucket name in the backup field. If you enter a full s3:// URI, Firehose nests it literally into the object key.

NOTE: The failed-record backup shares a bucket with the batch path's JSON files. The batch path's S3 event notification covers the whole bucket, so a file written under rt-failed/ after a failed delivery can also reach Snowpipe. In a healthy run nothing is written there. If you see unexpected rows in bronze_purchases, check the files under rt-failed/ first. For a longer-lived setup, use a separate bucket for the backup.

Build the Batch Path

Create the Storage Integration

CREATE OR REPLACE STORAGE INTEGRATION ticketing_s3_int
  TYPE                      = EXTERNAL_STAGE
  STORAGE_PROVIDER          = 'S3'
  ENABLED                   = TRUE
  STORAGE_AWS_ROLE_ARN      = 'arn:aws:iam::<aws-account-id>:role/PLACEHOLDER_WILL_UPDATE'
  STORAGE_ALLOWED_LOCATIONS = ('s3://<your-bucket-name>/');

DESC STORAGE INTEGRATION ticketing_s3_int;

Create the IAM Role

{
  "Version": "2012-10-17",
  "Statement": [
    { "Effect": "Allow", "Action": ["s3:GetObject"],  "Resource": "arn:aws:s3:::<your-bucket-name>/*" },
    { "Effect": "Allow", "Action": ["s3:ListBucket"], "Resource": "arn:aws:s3:::<your-bucket-name>" }
  ]
}

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Principal": { "AWS": "<snowflake-iam-user-arn>" },
      "Action": "sts:AssumeRole",
      "Condition": { "StringEquals": { "sts:ExternalId": "<storage-aws-external-id>" } }
    }
  ]
}

ALTER STORAGE INTEGRATION ticketing_s3_int
  SET STORAGE_AWS_ROLE_ARN = 'arn:aws:iam::<aws-account-id>:role/snowflake-ticketing-s3-role';

NOTE: The ARN must be a single-quoted string. An unquoted ARN is parsed as an identifier and causes a syntax error.

Create the External Stage

CREATE OR REPLACE STAGE ticketing_db.pipeline.ticketing_s3_stage
  URL                 = 's3://<your-bucket-name>/'
  STORAGE_INTEGRATION = ticketing_s3_int
  FILE_FORMAT         = (TYPE = 'JSON');

LIST @ticketing_db.pipeline.ticketing_s3_stage;

Create the SNS Topic

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "AllowS3Publish",
      "Effect": "Allow",
      "Principal": { "Service": "s3.amazonaws.com" },
      "Action": "SNS:Publish",
      "Resource": "arn:aws:sns:us-east-1:<aws-account-id>:ticketing-sns-topic",
      "Condition": { "ArnLike": { "aws:SourceArn": "arn:aws:s3:::<your-bucket-name>" } }
    },
    {
      "Sid": "AllowSnowflakeSubscribe",
      "Effect": "Allow",
      "Principal": { "AWS": "<snowflake-iam-user-arn>" },
      "Action": ["SNS:Subscribe", "SNS:Receive"],
      "Resource": "arn:aws:sns:us-east-1:<aws-account-id>:ticketing-sns-topic"
    }
  ]
}

NOTE: Always select all, delete, and paste when you replace a policy in the console. Partial edits cause brace mismatches that are hard to spot.

Send S3 Events to SNS

Create the Batch Firehose Stream

IMPORTANT: Keep the new line delimiter enabled. Without it, Firehose joins all the JSON records in a delivery into one line, and Snowpipe cannot parse them. Snowpipe expects one JSON object per line.

NOTE: The 0-second buffer cuts transit time sharply. It also writes many small files, roughly 60 to 100 for a full run, which is fine at this scale and is not advisable at production volume.

Create the Snowpipe

CREATE OR REPLACE PIPE ticketing_db.pipeline.ticketing_snowpipe
  AUTO_INGEST   = TRUE
  AWS_SNS_TOPIC = 'arn:aws:sns:us-east-1:<aws-account-id>:ticketing-sns-topic'
AS
COPY INTO ticketing_db.pipeline.bronze_purchases (
  purchase_id, event_id, seat_id, customer_id, price_paid,
  channel, promo_code, party_size, purchased_at, _src_file
)
FROM (
  SELECT
    $1:purchase_id::VARCHAR,
    $1:event_id::VARCHAR,
    $1:seat_id::VARCHAR,
    $1:customer_id::VARCHAR,
    $1:price_paid::NUMBER(10,2),
    $1:channel::VARCHAR,
    $1:promo_code::VARCHAR,
    $1:party_size::NUMBER,
    $1:purchased_at::TIMESTAMP_TZ,
    METADATA$FILENAME
  FROM @ticketing_db.pipeline.ticketing_s3_stage
)
FILE_FORMAT = (TYPE = 'JSON');

NOTE: A staged JSON file has no declared columns, so Snowflake loads each record as one VARIANT value in column position 1, written $1. $1:field_name extracts a key, and the ::TYPE suffix casts it.

NOTE: If Snowpipe does not pick up files, the SNS subscription may be stuck. CREATE OR REPLACE PIPE does not force a new subscription when one already exists, even a broken one. To reset, drop the pipe, delete the SNS topic and recreate it with the same name, reapply the access policy, check that the S3 event notification still points at the topic, then create the pipe again.

Create the Dynamic Tables

CREATE OR REPLACE DYNAMIC TABLE ticketing_db.pipeline.silver_purchases
    TARGET_LAG = '60 seconds'
    WAREHOUSE  = ticketing_wh
AS
SELECT
    purchase_id,
    seat_id,
    LEFT(seat_id, 1) AS section,
    customer_id,
    channel,
    promo_code,
    party_size,
    price_paid,
    price_paid * CASE promo_code
        WHEN 'EARLY20' THEN 0.80
        WHEN 'VIP10'   THEN 0.90
        WHEN 'GROUP15' THEN 0.85
        ELSE 1.0
    END AS effective_price,
    purchased_at,
    _loaded_at
FROM ticketing_db.pipeline.bronze_purchases;

CREATE OR REPLACE DYNAMIC TABLE ticketing_db.pipeline.gold_section_metrics
    TARGET_LAG = '60 seconds'
    WAREHOUSE  = ticketing_wh
AS
SELECT
    section,
    channel,
    COUNT(*)                         AS tickets_sold,
    ROUND(SUM(effective_price), 2)   AS total_revenue,
    ROUND(AVG(effective_price), 2)   AS avg_effective_price,
    ROUND(COUNT(*) / 20.0 * 100, 1)  AS sell_through_pct,   -- 20 seats per section
    MAX(purchased_at)                AS last_sale_at
FROM ticketing_db.pipeline.silver_purchases
GROUP BY section, channel;

NOTE: 60 seconds is the minimum target lag for a Dynamic Table. Gold is a plain GROUP BY, so a section appears only after its first sale. Right after a reset the table is empty, and the dashboard's metrics stay blank until the first purchases land.

Add Pipeline Controls

CREATE OR REPLACE PROCEDURE ticketing_db.pipeline.reset_demo()
RETURNS VARCHAR
LANGUAGE SQL
EXECUTE AS OWNER
AS
$$
BEGIN
    -- Reset data
    TRUNCATE TABLE ticketing_db.pipeline.bronze_purchases;
    TRUNCATE TABLE ticketing_db.pipeline.bronze_purchases_rt;

    UPDATE ticketing_db.pipeline.seats
    SET status = 'AVAILABLE', sold_at = NULL;

    -- Refresh Silver and Gold before suspending, so the dashboard shows
    -- an empty state right away instead of stale numbers.
    ALTER DYNAMIC TABLE ticketing_db.pipeline.silver_purchases REFRESH;
    ALTER DYNAMIC TABLE ticketing_db.pipeline.gold_section_metrics REFRESH;

    -- Suspend the pipeline to stop warehouse wake-ups
    ALTER DYNAMIC TABLE ticketing_db.pipeline.silver_purchases SUSPEND;
    ALTER DYNAMIC TABLE ticketing_db.pipeline.gold_section_metrics SUSPEND;
    ALTER TASK ticketing_db.pipeline.seat_status_update_task SUSPEND;

    RETURN 'Demo reset complete and pipeline suspended. Call resume_demo() to restart.';
END;
$$;

CREATE OR REPLACE PROCEDURE ticketing_db.pipeline.resume_demo()
RETURNS VARCHAR
LANGUAGE SQL
EXECUTE AS OWNER
AS
$$
BEGIN
    ALTER DYNAMIC TABLE ticketing_db.pipeline.silver_purchases RESUME;
    ALTER DYNAMIC TABLE ticketing_db.pipeline.gold_section_metrics RESUME;

    ALTER DYNAMIC TABLE ticketing_db.pipeline.silver_purchases REFRESH;
    ALTER DYNAMIC TABLE ticketing_db.pipeline.gold_section_metrics REFRESH;

    ALTER TASK ticketing_db.pipeline.seat_status_update_task RESUME;

    RETURN 'Pipeline resumed: dynamic tables refreshed and active, task running.';
END;
$$;

CALL ticketing_db.pipeline.reset_demo();    -- clears data, suspends everything
CALL ticketing_db.pipeline.resume_demo();   -- wakes everything up

IMPORTANT: reset_demo() leaves the pipeline suspended on purpose. It will not wake itself up, so always call resume_demo() before the next run.

If the Seat Map Stops Updating

SHOW TASKS LIKE 'SEAT_STATUS_UPDATE_TASK' IN SCHEMA ticketing_db.pipeline;   -- state should be "started"

SELECT name, state, error_code, error_message, scheduled_time
FROM TABLE(ticketing_db.INFORMATION_SCHEMA.TASK_HISTORY(
    TASK_NAME => 'SEAT_STATUS_UPDATE_TASK',
    SCHEDULED_TIME_RANGE_START => DATEADD('hour', -1, CURRENT_TIMESTAMP())))
ORDER BY scheduled_time DESC
LIMIT 20;

Launch the Dashboard

Download the Files

ticketing-pipeline/
  firehose-snowflake-auth/
    rsa_key.p8
    rsa_key.pub
  presentation-layer/
    dashboard.py
    .streamlit/
      secrets.toml
  ticket-generator/
    producer.py

Install the Packages

python -m venv venv

# macOS and Linux
source venv/bin/activate
# Windows
venv\Scripts\activate

pip install boto3 streamlit snowflake-snowpark-python cryptography plotly

Add Your Credentials

[snowflake]
account          = "<orgname>-<account_name>"
user             = "<your_user>"
private_key_path = "../firehose-snowflake-auth/rsa_key.p8"
warehouse        = "ticketing_wh"
database         = "ticketing_db"
schema           = "pipeline"
role             = "ACCOUNTADMIN"

IMPORTANT: Never commit secrets.toml. It names your account and the path to your private key.

Start the Dashboard

cd presentation-layer
streamlit run dashboard.py

NOTE: The video above is a recording of a complete live run. The architecture diagram shown in it is an earlier version. Where it differs from this guide, such as the phase names or the refresh rates, the guide is current.

Run the Producer

Wake the Pipeline

CALL ticketing_db.pipeline.resume_demo();

NOTE: The first file Snowpipe sees after a quiet period can take 25 to 30 seconds to load. If you want a smooth first run, send a small throwaway batch a minute ahead, then run reset_demo() and resume_demo() again.

Start Selling

cd ticket-generator
python producer.py

Verify the Pipeline

Check the Fast Path

SELECT COUNT(*) FROM ticketing_db.pipeline.bronze_purchases_rt;                    -- grows during the run

SELECT status, COUNT(*) FROM ticketing_db.pipeline.seats GROUP BY status;          -- SOLD count rises

SELECT * FROM ticketing_db.pipeline.seats
WHERE status = 'SOLD' ORDER BY sold_at DESC LIMIT 10;                             -- recent sold_at values

SELECT COUNT(*) FROM ticketing_db.pipeline.bronze_purchases_rt_stream;            -- near zero when the Task is caught up

Check the Batch Path

SELECT COUNT(*) FROM ticketing_db.pipeline.bronze_purchases;
SELECT COUNT(*) FROM ticketing_db.pipeline.silver_purchases;
SELECT * FROM ticketing_db.pipeline.gold_section_metrics ORDER BY section, channel LIMIT 10;

SELECT * FROM TABLE(INFORMATION_SCHEMA.COPY_HISTORY(
  TABLE_NAME => 'BRONZE_PURCHASES',
  START_TIME => DATEADD(HOUR, -1, CURRENT_TIMESTAMP())));

SHOW DYNAMIC TABLES IN SCHEMA ticketing_db.pipeline;

Troubleshooting

Clean Up

USE ROLE ACCOUNTADMIN;

ALTER TASK IF EXISTS ticketing_db.pipeline.seat_status_update_task SUSPEND;
DROP DATABASE IF EXISTS ticketing_db;
DROP STORAGE INTEGRATION IF EXISTS ticketing_s3_int;
DROP WAREHOUSE IF EXISTS ticketing_wh;
ALTER USER <your_user> UNSET RSA_PUBLIC_KEY;

Conclusion And Resources

What You Learned

Related Resources