Overview

What You'll Learn

What You'll Build

What You'll Need

Clone the Repository

git clone https://github.com/sfc-gh-cnantasenamat/build-real-time-bi-dashboard-interactive-warehouses.git
cd build-real-time-bi-dashboard-interactive-warehouses

├── scripts/
│   └── setup.sql             # Creates all Snowflake objects in one pass
├── spcs/
│   ├── Dockerfile            # Container definition for the simulation service
│   ├── requirements.txt      # Python dependencies (snowflake-connector-python)
│   ├── simulation_service.py # Long-running service loop
│   └── simulation_spec.yaml  # SPCS service specification
└── streamlit/
    ├── streamlit_app.py      # Live dashboard application
    └── environment.yml       # Streamlit in Snowflake Python environment

Create Snowflake Objects

Demo role

USE ROLE ACCOUNTADMIN;

CREATE ROLE IF NOT EXISTS INTERACTIVE_ANALYTICS_ROLE
  COMMENT = 'Owns all objects for the real-time BI dashboard demo';

GRANT CREATE DATABASE, CREATE WAREHOUSE, CREATE COMPUTE POOL,
      EXECUTE TASK, EXECUTE MANAGED TASK
  ON ACCOUNT TO ROLE INTERACTIVE_ANALYTICS_ROLE;

GRANT ROLE INTERACTIVE_ANALYTICS_ROLE TO ROLE SYSADMIN;

SET MY_USER = CURRENT_USER();
GRANT ROLE INTERACTIVE_ANALYTICS_ROLE TO USER IDENTIFIER($MY_USER);

Database and schema

USE ROLE INTERACTIVE_ANALYTICS_ROLE;

CREATE DATABASE IF NOT EXISTS INTERACTIVE_STREAMING_DEMO;
CREATE SCHEMA   IF NOT EXISTS INTERACTIVE_STREAMING_DEMO.PUBLIC;

Standard warehouse

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

USE WAREHOUSE INTERACTIVE_ANALYTICS_WH;

Orders table

CREATE OR REPLACE TABLE INTERACTIVE_STREAMING_DEMO.PUBLIC.ORDERS (
  ORDER_ID          STRING,
  ORDER_TS          TIMESTAMP_NTZ,
  REGION            STRING,
  PRODUCT_CATEGORY  STRING,
  QUANTITY          INT,
  AMOUNT            NUMBER(10,2)
)
CLUSTER BY (DATE_TRUNC('MINUTE', ORDER_TS), REGION);

Interactive warehouse

CREATE INTERACTIVE WAREHOUSE IF NOT EXISTS INTERACTIVE_ANALYTICS_IWH
  WAREHOUSE_SIZE      = 'XSMALL'
  AUTO_RESUME         = TRUE
  INITIALLY_SUSPENDED = TRUE;

ALTER WAREHOUSE INTERACTIVE_ANALYTICS_IWH RESUME IF SUSPENDED;

ALTER WAREHOUSE INTERACTIVE_ANALYTICS_IWH
  ADD TABLES (INTERACTIVE_STREAMING_DEMO.PUBLIC.ORDERS);

ALTER WAREHOUSE INTERACTIVE_ANALYTICS_IWH
  SET FALLBACK_WAREHOUSE = INTERACTIVE_ANALYTICS_WH;

-- Switch back: interactive warehouses cannot run INSERT, UPDATE or DELETE.
USE WAREHOUSE INTERACTIVE_ANALYTICS_WH;

Interactive warehouses enforce a 5-second query timeout and a 24-hour minimum auto-suspend. The FALLBACK_WAREHOUSE setting re-runs any query that exceeds the timeout on the standard warehouse instead of failing it, which mainly matters while the cache is still cold right after a resume. Keep dashboard queries selective (narrow projections, filters aligned with the clustering key) for consistent sub-second latency.

History and control tables

CREATE TABLE IF NOT EXISTS INTERACTIVE_STREAMING_DEMO.PUBLIC.ORDERS_HISTORY (
  ORDER_ID          STRING,
  ORDER_TS          TIMESTAMP_NTZ,
  REGION            STRING,
  PRODUCT_CATEGORY  STRING,
  QUANTITY          NUMBER,
  AMOUNT            NUMBER(10,2),
  TICK_AT           TIMESTAMP_NTZ DEFAULT CURRENT_TIMESTAMP()
);

CREATE OR REPLACE TABLE INTERACTIVE_STREAMING_DEMO.PUBLIC.SIMULATION_CONTROL (
  is_running      BOOLEAN       NOT NULL DEFAULT FALSE,
  mode            VARCHAR       NOT NULL DEFAULT 'normal',
  flash_category  VARCHAR,
  rate_mult       FLOAT         NOT NULL DEFAULT 1.0,
  last_active_at  TIMESTAMP_NTZ NOT NULL DEFAULT CURRENT_TIMESTAMP(),
  total_orders    NUMBER        NOT NULL DEFAULT 0
);

INSERT INTO INTERACTIVE_STREAMING_DEMO.PUBLIC.SIMULATION_CONTROL
  (is_running, mode, flash_category, rate_mult, last_active_at, total_orders)
VALUES (FALSE, 'normal', NULL, 1.0, CURRENT_TIMESTAMP(), 0);

Simulation stored procedure

CREATE OR REPLACE PROCEDURE INTERACTIVE_STREAMING_DEMO.PUBLIC.SP_SIMULATE_TICK(
    MODE           VARCHAR,
    FLASH_CATEGORY VARCHAR,
    NUM_ROWS       FLOAT,
    DOMINANT       VARCHAR,
    DOM_PCT        FLOAT
)
RETURNS VARCHAR
LANGUAGE JAVASCRIPT
EXECUTE AS OWNER
AS
$$
    var flash   = FLASH_CATEGORY || null;
    var numRows = Math.floor(NUM_ROWS);

    // ---- Tick categories ----
    // Pick 2-5 categories randomly for this tick.
    // If a Flash Sale is active, the promoted category is always included.
    var all_cats = ['Electronics','Apparel','Home','Beauty','Sports','Toys'];
    for (var i = all_cats.length - 1; i > 0; i--) {   // Fisher-Yates shuffle
        var j = Math.floor(Math.random() * (i + 1));
        var tmp = all_cats[i]; all_cats[i] = all_cats[j]; all_cats[j] = tmp;
    }
    var n_cats    = Math.floor(Math.random() * 4) + 2;
    var tick_cats = all_cats.slice(0, n_cats);
    if (MODE === 'flash_sale' && flash && tick_cats.indexOf(flash) === -1) {
        tick_cats[tick_cats.length - 1] = flash;
    }

    var n         = tick_cats.length;
    var quoted    = tick_cats.map(function(c) { return "'" + c + "'"; });
    var cat_array = 'ARRAY_CONSTRUCT(' + quoted.join(',') + ')';
    var cat_idx   = 'UNIFORM(0,' + (n - 1) + ',RANDOM())';

    var cat_sql, amount_sql;

    if (MODE === 'flash_sale' && flash) {
        // Flash Sale: 50 % of orders go to the promoted category; prices elevated.
        cat_sql    = "CASE WHEN UNIFORM(0,99,RANDOM()) < 50 THEN '" + flash + "'"
                   + " ELSE " + cat_array + "[" + cat_idx + "]::STRING END";
        amount_sql = "CASE WHEN PRODUCT_CATEGORY = '" + flash + "'"
                   + " THEN UNIFORM(50,800,RANDOM()) ELSE UNIFORM(5,400,RANDOM()) END";
    } else {
        // Each region has a preferred product category with a graduated
        // probability spectrum (28 % down to 8 %), creating a visible
        // diagonal gradient in the Region x Category heatmap.
        var rw = {
            'North America': {'Electronics':28,'Sports':22,'Beauty':18,'Home':14,'Toys':10,'Apparel':8},
            'Europe':        {'Apparel':28,'Electronics':22,'Beauty':20,'Toys':14,'Home':8,'Sports':8},
            'APAC':          {'Toys':28,'Home':22,'Electronics':18,'Beauty':14,'Apparel':10,'Sports':8},
            'LATAM':         {'Home':28,'Apparel':22,'Beauty':18,'Sports':14,'Toys':10,'Electronics':8},
            'MEA':           {'Beauty':28,'Toys':22,'Home':18,'Apparel':14,'Electronics':10,'Sports':8}
        };
        var regs   = ['North America','Europe','APAC','LATAM','MEA'];
        var rcases = [];
        for (var ri = 0; ri < regs.length; ri++) {
            var reg = regs[ri];
            var w   = rw[reg];
            var wl  = tick_cats.map(function(c) { return {cat: c, w: w[c] || 8}; });
            wl.sort(function(a, b) { return b.w - a.w; });
            var tot = 0;
            for (var wi = 0; wi < wl.length; wi++) { tot += wl[wi].w; }
            var cum = 0;
            var inn = [];
            for (var k = 0; k < wl.length - 1; k++) {
                cum += Math.round(wl[k].w * 100 / tot);
                inn.push('WHEN rv < ' + cum + " THEN '" + wl[k].cat + "'");
            }
            inn.push("ELSE '" + wl[wl.length - 1].cat + "'");
            // One UNIFORM draw per row via scalar subquery for accurate distribution.
            rcases.push(
                "WHEN REGION = '" + reg + "' THEN "
                + '(SELECT CASE ' + inn.join(' ') + ' END'
                + ' FROM (SELECT UNIFORM(0,99,RANDOM()) AS rv) _t)'
            );
        }
        rcases.push('ELSE ' + cat_array + '[' + cat_idx + ']::STRING');
        cat_sql    = 'CASE ' + rcases.join(' ') + ' END';
        amount_sql = 'UNIFORM(5,400,RANDOM())';
    }

    var sql = [
        'INSERT OVERWRITE INTO INTERACTIVE_STREAMING_DEMO.PUBLIC.ORDERS',
        'SELECT ORDER_ID, ORDER_TS, REGION, PRODUCT_CATEGORY, QUANTITY,',
        '       ROUND(UNIT_PRICE * QUANTITY, 2) AS AMOUNT',
        'FROM (',
        '  SELECT UUID_STRING() AS ORDER_ID,',
        "    DATEADD('second',-UNIFORM(0,300,RANDOM()),CURRENT_TIMESTAMP())",
        '      ::TIMESTAMP_NTZ AS ORDER_TS,',
        "    ARRAY_CONSTRUCT('North America','Europe','APAC','LATAM','MEA')",
        '      [UNIFORM(0,4,RANDOM())]::STRING AS REGION,',
        '    ' + cat_sql + ' AS PRODUCT_CATEGORY,',
        '    UNIFORM(1,5,RANDOM()) AS QUANTITY,',
        '    ' + amount_sql + ' AS UNIT_PRICE',
        '  FROM TABLE(GENERATOR(ROWCOUNT => ' + numRows + '))',
        ')'
    ];
    snowflake.execute({sqlText: sql.join(' ')});

    snowflake.execute({sqlText:
        'INSERT INTO INTERACTIVE_STREAMING_DEMO.PUBLIC.ORDERS_HISTORY'
        + ' (ORDER_ID, ORDER_TS, REGION, PRODUCT_CATEGORY, QUANTITY, AMOUNT, TICK_AT)'
        + ' SELECT ORDER_ID, ORDER_TS, REGION, PRODUCT_CATEGORY, QUANTITY, AMOUNT,'
        + ' CURRENT_TIMESTAMP() FROM INTERACTIVE_STREAMING_DEMO.PUBLIC.ORDERS'
    });

    snowflake.execute({sqlText:
        'UPDATE INTERACTIVE_STREAMING_DEMO.PUBLIC.SIMULATION_CONTROL'
        + ' SET TOTAL_ORDERS = COALESCE(TOTAL_ORDERS, 0) + ' + numRows
        + ',    LAST_ACTIVE_AT = CURRENT_TIMESTAMP()'
    });
    return 'OK';
$$;

Compute pool and image repository

CREATE COMPUTE POOL IF NOT EXISTS INTERACTIVE_ANALYTICS_POOL
  MIN_NODES         = 1
  MAX_NODES         = 1
  INSTANCE_FAMILY   = CPU_X64_XS
  AUTO_RESUME       = TRUE
  AUTO_SUSPEND_SECS = 3600;

CREATE IMAGE REPOSITORY IF NOT EXISTS INTERACTIVE_STREAMING_DEMO.PUBLIC.SIM_REPO;

-- Note the repository_url in the output; you will use it in the next step.
SHOW IMAGE REPOSITORIES LIKE 'SIM_REPO'
  IN SCHEMA INTERACTIVE_STREAMING_DEMO.PUBLIC;

App stage

CREATE STAGE IF NOT EXISTS INTERACTIVE_STREAMING_DEMO.PUBLIC.APP_STAGE;

Maintenance tasks

CREATE OR REPLACE TASK INTERACTIVE_STREAMING_DEMO.PUBLIC.INTERACTIVE_ANALYTICS_AUTOSTOP
  SCHEDULE                                 = '15 MINUTE'
  USER_TASK_MANAGED_INITIAL_WAREHOUSE_SIZE = 'XSMALL'
  COMMENT                                  = 'Auto-stop simulation when idle > 1 h'
AS
  UPDATE INTERACTIVE_STREAMING_DEMO.PUBLIC.SIMULATION_CONTROL
  SET is_running = FALSE
  WHERE is_running = TRUE
    AND last_active_at < DATEADD('hour', -1, CURRENT_TIMESTAMP());

CREATE OR REPLACE TASK INTERACTIVE_STREAMING_DEMO.PUBLIC.INTERACTIVE_ANALYTICS_HISTORY_CLEANUP
  WAREHOUSE = INTERACTIVE_ANALYTICS_WH
  SCHEDULE  = 'USING CRON 0 2 * * * UTC'
  COMMENT   = 'Delete ORDERS_HISTORY rows older than 1 day'
AS
  DELETE FROM INTERACTIVE_STREAMING_DEMO.PUBLIC.ORDERS_HISTORY
  WHERE TICK_AT < DATEADD('day', -1, CURRENT_TIMESTAMP());

ALTER TASK INTERACTIVE_STREAMING_DEMO.PUBLIC.INTERACTIVE_ANALYTICS_AUTOSTOP RESUME;
ALTER TASK INTERACTIVE_STREAMING_DEMO.PUBLIC.INTERACTIVE_ANALYTICS_HISTORY_CLEANUP RESUME;

Seed initial orders

INSERT OVERWRITE INTO INTERACTIVE_STREAMING_DEMO.PUBLIC.ORDERS
SELECT
  UUID_STRING()                                                       AS ORDER_ID,
  DATEADD('second', -UNIFORM(0, 1800, RANDOM()), CURRENT_TIMESTAMP())
    ::TIMESTAMP_NTZ                                                   AS ORDER_TS,
  ARRAY_CONSTRUCT('North America','Europe','APAC','LATAM','MEA')
    [UNIFORM(0, 4, RANDOM())]::STRING                                 AS REGION,
  ARRAY_CONSTRUCT('Electronics','Apparel','Home','Beauty','Sports','Toys')
    [UNIFORM(0, 5, RANDOM())]::STRING                                 AS PRODUCT_CATEGORY,
  UNIFORM(1, 5, RANDOM())                                             AS QUANTITY,
  ROUND(UNIFORM(5, 400, RANDOM()) * UNIFORM(1, 5, RANDOM()), 2)      AS AMOUNT
FROM TABLE(GENERATOR(ROWCOUNT => 2000));

Verify the objects

Build the Container

How the service authenticates

# spcs/simulation_service.py (key excerpt)
WAREHOUSE = os.environ.get("SNOWFLAKE_WAREHOUSE", "INTERACTIVE_ANALYTICS_WH")
ROLE = os.environ.get("SNOWFLAKE_ROLE", "INTERACTIVE_ANALYTICS_ROLE")
TOKEN_FILE = "/snowflake/session/token"

def connect():
    with open(TOKEN_FILE) as f:
        token = f.read().strip()
    return snowflake.connector.connect(
        host=os.environ["SNOWFLAKE_HOST"],
        account=os.environ["SNOWFLAKE_ACCOUNT"],
        authenticator="oauth",
        token=token,
        database="INTERACTIVE_STREAMING_DEMO",
        schema="PUBLIC",
        role=ROLE,
        warehouse=WAREHOUSE,
    )

Get the registry URL

SHOW IMAGE REPOSITORIES LIKE 'SIM_REPO'
  IN SCHEMA INTERACTIVE_STREAMING_DEMO.PUBLIC;

Log in to the registry

docker login <your_registry_url>

If you have the Snowflake CLI installed and a connection configured, snow spcs image-registry login logs Docker in to your account's registry using that connection, so you don't have to type a password.

Build, tag, and push the image

docker build -t simulation ./spcs/

docker tag simulation \
  <your_registry_url>/simulation:latest

docker push <your_registry_url>/simulation:latest

Building on Apple Silicon (arm64)? Add --platform linux/amd64 to docker build so the image runs correctly on Snowflake's x86_64 compute pool nodes.

Deploy Everything

Create the SPCS simulation service

USE ROLE INTERACTIVE_ANALYTICS_ROLE;
USE WAREHOUSE INTERACTIVE_ANALYTICS_WH;

CREATE SERVICE IF NOT EXISTS
  INTERACTIVE_STREAMING_DEMO.PUBLIC.INTERACTIVE_ANALYTICS_SVC
  IN COMPUTE POOL INTERACTIVE_ANALYTICS_POOL
  FROM SPECIFICATION $$
  spec:
    container:
    - name: simulator
      image: /interactive_streaming_demo/public/sim_repo/simulation:latest
      env:
        SNOWFLAKE_WAREHOUSE: INTERACTIVE_ANALYTICS_WH
        SIM_INTERVAL: "1"
        IDLE_SUSPEND_SECS: "3600"
        ROWS_BASE: "100"
        CTRL_CACHE_TICKS: "5"
  $$
  MIN_INSTANCES = 1
  MAX_INSTANCES = 1;

SHOW SERVICES LIKE 'INTERACTIVE_ANALYTICS_SVC'
  IN SCHEMA INTERACTIVE_STREAMING_DEMO.PUBLIC;

SELECT SYSTEM$GET_SERVICE_LOGS(
  'INTERACTIVE_STREAMING_DEMO.PUBLIC.INTERACTIVE_ANALYTICS_SVC',
  '0', 'simulator', 50
);

INFO Simulation service starting (interval=1.0s, idle_suspend=3600s, rows_base=100, cache_ticks=5)
INFO Connecting to GLOBAL Snowflake domain
INFO Connected to Snowflake as INTERACTIVE_ANALYTICS_ROLE
INFO Idle timer reset

Upload the Streamlit dashboard

USE ROLE INTERACTIVE_ANALYTICS_ROLE;

PUT file://streamlit_app.py
    @INTERACTIVE_STREAMING_DEMO.PUBLIC.APP_STAGE/streamlit/
    OVERWRITE=TRUE AUTO_COMPRESS=FALSE;

PUT file://environment.yml
    @INTERACTIVE_STREAMING_DEMO.PUBLIC.APP_STAGE/streamlit/
    OVERWRITE=TRUE AUTO_COMPRESS=FALSE;

CREATE OR REPLACE STREAMLIT INTERACTIVE_STREAMING_DEMO.PUBLIC.LIVE_ORDERS_DASHBOARD
  ROOT_LOCATION   = '@INTERACTIVE_STREAMING_DEMO.PUBLIC.APP_STAGE/streamlit'
  MAIN_FILE       = 'streamlit_app.py'
  QUERY_WAREHOUSE = INTERACTIVE_ANALYTICS_WH
  TITLE           = 'Live Orders Dashboard';

Explore the Dashboard

Start the simulation

KPI tiles

Adjust settings

Flash Sale

Region x Category heatmap

New session

Clean Up

Warning: Interactive warehouses have a 24-hour minimum billing period once created. Drop the warehouse as soon as you are done to avoid unexpected charges.

USE ROLE INTERACTIVE_ANALYTICS_ROLE;
USE WAREHOUSE INTERACTIVE_ANALYTICS_WH;

-- Stop the data feed before dropping objects.
UPDATE INTERACTIVE_STREAMING_DEMO.PUBLIC.SIMULATION_CONTROL
  SET is_running = FALSE;

ALTER TASK INTERACTIVE_STREAMING_DEMO.PUBLIC.INTERACTIVE_ANALYTICS_AUTOSTOP SUSPEND;
ALTER TASK INTERACTIVE_STREAMING_DEMO.PUBLIC.INTERACTIVE_ANALYTICS_HISTORY_CLEANUP SUSPEND;

DROP SERVICE      IF EXISTS INTERACTIVE_STREAMING_DEMO.PUBLIC.INTERACTIVE_ANALYTICS_SVC;
DROP STREAMLIT    IF EXISTS INTERACTIVE_STREAMING_DEMO.PUBLIC.LIVE_ORDERS_DASHBOARD;
DROP TASK         IF EXISTS INTERACTIVE_STREAMING_DEMO.PUBLIC.INTERACTIVE_ANALYTICS_AUTOSTOP;
DROP TASK         IF EXISTS INTERACTIVE_STREAMING_DEMO.PUBLIC.INTERACTIVE_ANALYTICS_HISTORY_CLEANUP;
DROP PROCEDURE    IF EXISTS INTERACTIVE_STREAMING_DEMO.PUBLIC.SP_SIMULATE_TICK(VARCHAR, VARCHAR, FLOAT, VARCHAR, FLOAT);
DROP WAREHOUSE    IF EXISTS INTERACTIVE_ANALYTICS_IWH;
DROP WAREHOUSE    IF EXISTS INTERACTIVE_ANALYTICS_WH;
DROP COMPUTE POOL IF EXISTS INTERACTIVE_ANALYTICS_POOL;
DROP DATABASE     IF EXISTS INTERACTIVE_STREAMING_DEMO;

-- Remove the demo role last.
USE ROLE ACCOUNTADMIN;
DROP ROLE         IF EXISTS INTERACTIVE_ANALYTICS_ROLE;

Conclusion And Resources

What You Learned

Related Resources