Overview

Note: Code Bundles are in Public Preview.

What You'll Learn

What You'll Need

Tip: You can run this Quickstart with Cortex Code, Snowflake's AI coding assistant. Point Cortex Code at this guide and it can execute each step for you, explain what's happening, and help you adapt the pipeline to your own data. You can also use it side-by-side as you work through the guide in your terminal.

Architecture And Project Setup

The Clickstream Pipeline

Raw Events (~50K)           SESSIONS (~12K)                FUNNEL_METRICS (8 rows)
(EVENTS table)             (SESSIONS table)               (FUNNEL_METRICS table)
[page_view, cart, ...]  -->  [session_id, user_id,  -->    [step, event_name,
                             duration, revenue, ...]        unique_users, ...]

                             ^^^^^^^^^^^^^^^^^^^^^^        ^^^^^^^^^^^^^^^^^^^^^^^
                             Stage 1: Python               Stage 2: PySpark
                             (warehouse compute)           (serverless Spark)

Clone the Companion Repository

git clone https://github.com/Snowflake-Labs/sfguide-run-python-and-spark-jobs-with-code-bundles.git
cd sfguide-run-python-and-spark-jobs-with-code-bundles

sfguide-run-python-and-spark-jobs-with-code-bundles/
├── setup.sql                 # Database, warehouse, stage, synthetic data
├── teardown.sql              # Clean up all created objects
├── snowflake.yml             # Snowflake CLI project definition (with code_bundle)
└── src/
    ├── bundle.yml             # Bundle specification (entrypoints, env, warehouse)
    ├── requirements.txt       # Dependencies installed into execution environment
    ├── sessionize.py          # Python sessionization job (runs on warehouse)
    ├── analytics.py           # PySpark funnel analysis job (runs on Spark engine)
    └── helpers.py             # Shared session + logging utilities

Environment Setup

Run setup.sql

snow sql -f setup.sql

-- Create database, schema, and warehouse
CREATE OR REPLACE DATABASE CODE_BUNDLES_QUICKSTART;
CREATE OR REPLACE SCHEMA CODE_BUNDLES_QUICKSTART.CLICKSTREAM;

CREATE OR REPLACE WAREHOUSE CODE_BUNDLES_WH
  WAREHOUSE_SIZE = 'X-SMALL'
  AUTO_SUSPEND = 60
  AUTO_RESUME = TRUE;

USE DATABASE CODE_BUNDLES_QUICKSTART;
USE SCHEMA CLICKSTREAM;
USE WAREHOUSE CODE_BUNDLES_WH;

-- Stage for the code bundle artifacts
CREATE OR REPLACE STAGE BUNDLE_STAGE
  DIRECTORY = (ENABLE = TRUE);

-- Generate ~50K synthetic clickstream events across 2,500 users
CREATE OR REPLACE TABLE EVENTS AS
WITH RECURSIVE
  users AS (
    SELECT
      SEQ4() AS user_id,
      'user_' || LPAD(SEQ4()::VARCHAR, 6, '0') AS user_handle,
      CASE MOD(SEQ4(), 3)
        WHEN 0 THEN 'mobile'
        WHEN 1 THEN 'desktop'
        ELSE 'tablet'
      END AS device_type,
      CASE MOD(SEQ4(), 4)
        WHEN 0 THEN 'direct'
        WHEN 1 THEN 'organic_search'
        WHEN 2 THEN 'paid_ad'
        ELSE 'social'
      END AS traffic_source
    FROM TABLE(GENERATOR(ROWCOUNT => 2500))
  ),
  event_seq AS (
    SELECT
      SEQ4() AS event_idx,
      MOD(ABS(RANDOM()), 2500) AS user_id,
      -- Spread events over a 7-day window
      TIMESTAMPADD(
        SECOND,
        MOD(ABS(RANDOM()), 7 * 86400),
        '2025-01-01 00:00:00'::TIMESTAMP_NTZ
      ) AS event_time,
      UNIFORM(1, 100, RANDOM()) AS event_roll,
      ROUND(UNIFORM(500, 25000, RANDOM()) / 100.0, 2) AS price
    FROM TABLE(GENERATOR(ROWCOUNT => 50000))
  )
SELECT
  UUID_STRING() AS event_id,
  u.user_handle AS user_id,
  e.event_time,
  CASE
    WHEN e.event_roll <= 50 THEN 'page_view'
    WHEN e.event_roll <= 75 THEN 'product_detail'
    WHEN e.event_roll <= 90 THEN 'add_to_cart'
    WHEN e.event_roll <= 97 THEN 'checkout_start'
    ELSE 'purchase'
  END AS event_name,
  CASE
    WHEN e.event_roll > 75 THEN e.price
    ELSE NULL
  END AS amount,
  u.device_type,
  u.traffic_source
FROM event_seq e
JOIN users u ON e.user_id = u.user_id
ORDER BY e.event_time;

SELECT event_name, COUNT(*) AS event_count, ROUND(AVG(amount), 2) AS avg_amount
FROM EVENTS
GROUP BY event_name
ORDER BY event_count DESC;

Anatomy Of A Code Bundle

snowflake.yml

definition_version: '2'
entities:
  clickstream_bundle:
    type: code_bundle
    identifier:
      name: CLICKSTREAM_BUNDLE
      database: CODE_BUNDLES_QUICKSTART
      schema: CLICKSTREAM
    stage: BUNDLE_STAGE
    spec: bundle.yml
    artifacts:
      - src/*

src/bundle.yml

spec_version: 1.0

default:
  entrypoint: sessionize.py
  compute:
    warehouse: CODE_BUNDLES_WH
  runtime:
    language: python
    version: '3.11'
    dependencies:
      requirements: requirements.txt

entrypoints:
  sessionize:
    file: sessionize.py
    compute:
      warehouse: CODE_BUNDLES_WH
    runtime:
      language: python
      version: '3.11'
      dependencies:
        requirements: requirements.txt

  analytics:
    file: analytics.py
    runtime:
      language: python
      version: '3.11'
      dependencies:
        requirements: requirements.txt

Inspect the Code

# sessionize.py
from helpers import get_session, log_step

def main() -> None:
    args = parse_args()
    session = get_session()

    log_step(f"Sessionizing {args.source_table} (timeout: {args.inactivity_minutes} min)")
    
    session.sql(f"""
        -- Sessionization with window-based inactivity timeout
        ... 
    """).collect()

# analytics.py
from pyspark.sql import functions as F
from helpers import get_spark_session, log_step

def main() -> None:
    spark = get_spark_session()
    log_step("Running funnel analysis with PySpark...")

    df = spark.table("CODE_BUNDLES_QUICKSTART.CLICKSTREAM.SESSIONS")
    
    # Funnel analysis using PySpark DataFrame operations
    funnel = df.select(
        F.countDistinct("USER_ID").alias("total_users"),
        F.countDistinct(F.when(F.col("PAGE_VIEWS") > 0, F.col("USER_ID"))).alias("viewed_page"),
        F.countDistinct(F.when(F.col("CART_ADDS") > 0, F.col("USER_ID"))).alias("added_to_cart"),
        F.countDistinct(F.when(F.col("PURCHASES") > 0, F.col("USER_ID"))).alias("purchased"),
    )
    
    funnel.write.mode("overwrite").saveAsTable(
        "CODE_BUNDLES_QUICKSTART.CLICKSTREAM.FUNNEL_METRICS"
    )

Deploy The Code Bundle

Deploy with snow code-bundle

snow code-bundle deploy

Uploading artifacts to @CODE_BUNDLES_QUICKSTART.CLICKSTREAM.BUNDLE_STAGE/CLICKSTREAM_BUNDLE/...
Creating or updating code bundle CLICKSTREAM_BUNDLE...
Code bundle CLICKSTREAM_BUNDLE successfully deployed.

What Just Happened?

USE DATABASE CODE_BUNDLES_QUICKSTART;
USE SCHEMA CLICKSTREAM;

SHOW CODE BUNDLES;

DESCRIBE CODE BUNDLE CLICKSTREAM_BUNDLE;

Run Python Jobs On Warehouse Compute

Run the Default Entrypoint

USE DATABASE CODE_BUNDLES_QUICKSTART;
USE SCHEMA CLICKSTREAM;
USE WAREHOUSE CODE_BUNDLES_WH;

EXECUTE CODE BUNDLE CLICKSTREAM_BUNDLE;

snow sql -q "EXECUTE CODE BUNDLE CODE_BUNDLES_QUICKSTART.CLICKSTREAM.CLICKSTREAM_BUNDLE;"

SELECT
  COUNT(*) AS total_sessions,
  COUNT(DISTINCT user_id) AS unique_users,
  ROUND(AVG(duration_minutes), 1) AS avg_duration_min,
  ROUND(AVG(page_views), 1) AS avg_page_views,
  SUM(has_purchase) AS sessions_with_purchase,
  ROUND(SUM(total_revenue), 2) AS total_revenue
FROM SESSIONS;

Run a Named Entrypoint

EXECUTE CODE BUNDLE CLICKSTREAM_BUNDLE
  ENTRYPOINT = 'sessionize';

Pass CLI Arguments to the Script

EXECUTE CODE BUNDLE CLICKSTREAM_BUNDLE
  ENTRYPOINT = 'sessionize'
  ARGS = ('--inactivity-minutes', '45', '--target-table', 'SESSIONS_45MIN');

SELECT
  '30-min timeout' AS run_type,
  COUNT(*) AS total_sessions,
  ROUND(AVG(duration_minutes), 1) AS avg_duration_min
FROM SESSIONS
UNION ALL
SELECT
  '45-min timeout' AS run_type,
  COUNT(*) AS total_sessions,
  ROUND(AVG(duration_minutes), 1) AS avg_duration_min
FROM SESSIONS_45MIN;

Override the Specification at Runtime

EXECUTE CODE BUNDLE CLICKSTREAM_BUNDLE
  ENTRYPOINT = 'sessionize'
  WITH SPECIFICATION = $$
    compute:
      warehouse: CODE_BUNDLES_WH
    env:
      LOG_LEVEL: DEBUG
  $$;

Run PySpark Jobs With No Spark Cluster

Submit via the Spark REST API

export SNOWFLAKE_TOKEN=$(snow connection test --format json | python3 -c "import sys, json; print(json.load(sys.stdin).get('token', ''))")

export SNOWFLAKE_ACCOUNT=$(snow connection test --format json | python3 -c "import sys, json; print(json.load(sys.stdin).get('account', ''))")

curl -X POST \
  "https://${SNOWFLAKE_ACCOUNT}.snowflakecomputing.com/api/v2/spark/jobs" \
  -H "Authorization: Bearer ${SNOWFLAKE_TOKEN}" \
  -H "Content-Type: application/json" \
  -d '{
    "mainApplicationFile": "@CODE_BUNDLES_QUICKSTART.CLICKSTREAM.BUNDLE_STAGE/CLICKSTREAM_BUNDLE/analytics.py",
    "appArgs": [
      "--source-table", "CODE_BUNDLES_QUICKSTART.CLICKSTREAM.SESSIONS",
      "--target-table", "CODE_BUNDLES_QUICKSTART.CLICKSTREAM.FUNNEL_METRICS"
    ],
    "sparkProperties": {
      "spark.snowflake.database": "CODE_BUNDLES_QUICKSTART",
      "spark.snowflake.schema": "CLICKSTREAM"
    }
  }'

{
  "jobId": "spark-job-01b4c8a2-7e3f-4a1d-9e0a-123456789abc",
  "status": "SUBMITTED"
}

Monitor the Spark Job

curl -X GET \
  "https://${SNOWFLAKE_ACCOUNT}.snowflakecomputing.com/api/v2/spark/jobs/<JOB_ID>" \
  -H "Authorization: Bearer ${SNOWFLAKE_TOKEN}"

Run the Spark Entrypoint via SQL

USE DATABASE CODE_BUNDLES_QUICKSTART;
USE SCHEMA CLICKSTREAM;

EXECUTE CODE BUNDLE CLICKSTREAM_BUNDLE
  ENTRYPOINT = 'analytics';

Verify Funnel Metrics

SELECT
  step_number,
  step_name,
  unique_users,
  conversion_rate_pct,
  dropoff_rate_pct
FROM FUNNEL_METRICS
ORDER BY step_number ASC;

+-------------+------------------+--------------+---------------------+------------------+
| STEP_NUMBER | STEP_NAME        | UNIQUE_USERS | CONVERSION_RATE_PCT | DROPOFF_RATE_PCT |
+-------------+------------------+--------------+---------------------+------------------+
|           1 | Homepage / Land  |         2500 |              100.00 |             0.00 |
|           2 | Product Detail   |         2180 |               87.20 |            12.80 |
|           3 | Add to Cart      |         1425 |               57.00 |            34.63 |
|           4 | Checkout Start   |          710 |               28.40 |            50.18 |
|           5 | Purchase         |          380 |               15.20 |            46.48 |
+-------------+------------------+--------------+---------------------+------------------+

Orchestrate With Snowflake Tasks

Create the Task Graph

USE DATABASE CODE_BUNDLES_QUICKSTART;
USE SCHEMA CLICKSTREAM;
USE WAREHOUSE CODE_BUNDLES_WH;

-- Root task: Run Python sessionizer on a daily schedule
CREATE OR REPLACE TASK CLICKSTREAM_SESSIONIZE_TASK
  WAREHOUSE = CODE_BUNDLES_WH
  SCHEDULE = 'USING CRON 0 2 * * * UTC'  -- Daily at 2:00 AM UTC
AS
  EXECUTE CODE BUNDLE CLICKSTREAM_BUNDLE
    ENTRYPOINT = 'sessionize';

-- Child task: Run PySpark funnel analysis after sessionization completes
CREATE OR REPLACE TASK CLICKSTREAM_ANALYTICS_TASK
  AFTER CLICKSTREAM_SESSIONIZE_TASK
AS
  EXECUTE CODE BUNDLE CLICKSTREAM_BUNDLE
    ENTRYPOINT = 'analytics';

Test the Pipeline Manually

-- Child tasks must be resumed before root tasks
ALTER TASK CLICKSTREAM_ANALYTICS_TASK RESUME;
ALTER TASK CLICKSTREAM_SESSIONIZE_TASK RESUME;

-- Trigger an immediate manual run of the root task
EXECUTE TASK CLICKSTREAM_SESSIONIZE_TASK;

Monitor Task Execution

SELECT
  name,
  state,
  scheduled_time,
  completed_time,
  TIMESTAMPDIFF('second', scheduled_time, completed_time) AS duration_seconds,
  error_message
FROM TABLE(INFORMATION_SCHEMA.TASK_HISTORY(
  TASK_NAME => 'CLICKSTREAM_SESSIONIZE_TASK',
  SCHEDULED_TIME_RANGE_START => DATEADD('hour', -1, CURRENT_TIMESTAMP())
))
ORDER BY scheduled_time DESC;

SELECT
  name,
  state,
  scheduled_time,
  completed_time,
  TIMESTAMPDIFF('second', scheduled_time, completed_time) AS duration_seconds,
  error_message
FROM TABLE(INFORMATION_SCHEMA.TASK_HISTORY(
  TASK_NAME => 'CLICKSTREAM_ANALYTICS_TASK',
  SCHEDULED_TIME_RANGE_START => DATEADD('hour', -1, CURRENT_TIMESTAMP())
))
ORDER BY scheduled_time DESC;

Production Patterns: Versioning, Async, Monitoring

Pattern 1: Versioning and Aliases

# Deploy a new version
snow code-bundle deploy --version v1.1.0

# Or deploy with an alias
snow code-bundle deploy --version v1.1.0 --alias prod

-- Execute a specific version
EXECUTE CODE BUNDLE CLICKSTREAM_BUNDLE
  VERSION = 'v1.1.0';

-- Execute via alias
EXECUTE CODE BUNDLE CLICKSTREAM_BUNDLE
  ALIAS = 'prod';

-- List registered versions
SHOW VERSIONS IN CODE BUNDLE CLICKSTREAM_BUNDLE;

-- Point the prod alias to a different version (instant rollback)
ALTER CODE BUNDLE CLICKSTREAM_BUNDLE
  SET ALIAS prod = 'v1.0.0';

-- Drop an old version
ALTER CODE BUNDLE CLICKSTREAM_BUNDLE
  DROP VERSION 'v0.9.0';

Pattern 2: Asynchronous Execution and Status Checking

snow sql -q "EXECUTE CODE BUNDLE CODE_BUNDLES_QUICKSTART.CLICKSTREAM.CLICKSTREAM_BUNDLE;" --async

Query ID: 01b4c8a2-0001-2345-0000-123456789abc
Status: RUNNING

SELECT
  query_id,
  query_text,
  execution_status,
  warehouse_name,
  start_time,
  end_time,
  total_elapsed_time / 1000 AS elapsed_seconds,
  error_code,
  error_message
FROM TABLE(INFORMATION_SCHEMA.QUERY_HISTORY())
WHERE query_id = '01b4c8a2-0001-2345-0000-123456789abc';

Pattern 3: Centralized Monitoring and Query History

-- Find all Code Bundle executions in the last 24 hours
SELECT
  query_id,
  user_name,
  warehouse_name,
  execution_status,
  start_time,
  end_time,
  ROUND(total_elapsed_time / 1000.0, 1) AS elapsed_sec,
  ROUND(compilation_time / 1000.0, 1) AS compile_sec,
  ROUND(execution_time / 1000.0, 1) AS exec_sec
FROM TABLE(INFORMATION_SCHEMA.QUERY_HISTORY(
  END_TIME_RANGE_START => DATEADD('hour', -24, CURRENT_TIMESTAMP())
))
WHERE query_text ILIKE '%EXECUTE CODE BUNDLE%'
ORDER BY start_time DESC;

Clean Up

snow sql -f teardown.sql

-- Suspend tasks first
ALTER TASK IF EXISTS CLICKSTREAM_SESSIONIZE_TASK SUSPEND;
ALTER TASK IF EXISTS CLICKSTREAM_ANALYTICS_TASK SUSPEND;

-- Drop all objects
DROP DATABASE IF EXISTS CODE_BUNDLES_QUICKSTART;
DROP WAREHOUSE IF EXISTS CODE_BUNDLES_WH;

Conclusion And Resources

What You Learned

Related Resources