Snowflake: Data Warehousing, Snowpark & Data Sharing

┌──────────────────────────────────────────────────────────────┐
│                     Snowflake Architecture                    │
│                                                              │
│  ┌────────────────────────────────────────────────────────┐  │
│  │              Query Layer (Cloud Services)               │  │
│  │  Optimizer, metadata, auth, transactions, security      │  │
│  └────────────────────────────────────────────────────────┘  │
│                                                              │
│  ┌────────────────────────────────────────────────────────┐  │
│  │              Compute Layer (Virtual Warehouses)         │  │
│  │  XS / S / M / L / XL — independent clusters           │  │
│  │  Auto-suspend + auto-resume — pay only when active     │  │
│  └────────────────────────────────────────────────────────┘  │
│                                                              │
│  ┌────────────────────────────────────────────────────────┐  │
│  │              Storage Layer (Micro-partitions)           │  │
│  │  Columnar, compressed Parquet — S3/Azure Blob/GCS      │  │
│  │  Automatic clustering, pruning, metadata indexing      │  │
│  └────────────────────────────────────────────────────────┘  │
└──────────────────────────────────────────────────────────────┘

-- Create a warehouse with auto-suspend
CREATE WAREHOUSE analytics_wh
    WAREHOUSE_SIZE = 'MEDIUM'
    AUTO_SUSPEND = 300        -- suspend after 5 min idle
    AUTO_RESUME = TRUE        -- resume automatically on query
    MAX_CLUSTER_COUNT = 3     -- multi-cluster: scale out under concurrency
    MIN_CLUSTER_COUNT = 1
    SCALING_POLICY = 'STANDARD';

-- Switch warehouse in session
USE WAREHOUSE analytics_wh;

-- Monitor warehouse credit usage
SELECT warehouse_name,
       SUM(credits_used) AS total_credits,
       COUNT(*)           AS query_count,
       AVG(execution_time)/1000 AS avg_sec
FROM snowflake.account_usage.warehouse_metering_history
WHERE start_time >= DATEADD(day, -7, CURRENT_TIMESTAMP())
GROUP BY 1
ORDER BY total_credits DESC;

-- Three-level namespace: database.schema.object
CREATE DATABASE analytics_prod;
CREATE SCHEMA analytics_prod.sales;

-- Table
CREATE OR REPLACE TABLE analytics_prod.sales.orders (
    order_id    VARCHAR(36)    NOT NULL,
    customer_id VARCHAR(36)    NOT NULL,
    amount      NUMBER(12, 2)  NOT NULL,
    status      VARCHAR(20)    DEFAULT 'pending',
    created_at  TIMESTAMP_NTZ  DEFAULT CURRENT_TIMESTAMP(),
    CONSTRAINT pk_orders PRIMARY KEY (order_id)
);

-- Clustering key: improves pruning on large tables queried by a specific column
ALTER TABLE analytics_prod.sales.orders
    CLUSTER BY (DATE_TRUNC('month', created_at));

-- VARIANT column for flexible JSON
CREATE TABLE events (
    event_id   VARCHAR(36),
    event_type VARCHAR(100),
    payload    VARIANT,        -- stores raw JSON
    received_at TIMESTAMP_NTZ DEFAULT CURRENT_TIMESTAMP()
);

-- Insert JSON
INSERT INTO events (event_id, event_type, payload)
SELECT UUID_STRING(),
       'order_placed',
       PARSE_JSON('{
           "order_id": "ORD-001",
           "customer": {"id": "C-123", "name": "Alice"},
           "items": [{"sku": "SKU-1", "qty": 2, "price": 29.99}]
       }');

-- Query JSON with : and [] notation
SELECT
    payload:order_id::VARCHAR              AS order_id,
    payload:customer.name::VARCHAR         AS customer_name,
    payload:items[0].price::FLOAT          AS first_item_price,
    ARRAY_SIZE(payload:items)              AS item_count
FROM events
WHERE event_type = 'order_placed';

-- FLATTEN: expand array elements into rows
SELECT
    e.event_id,
    item.value:sku::VARCHAR    AS sku,
    item.value:qty::INT        AS quantity,
    item.value:price::FLOAT    AS unit_price
FROM events e,
LATERAL FLATTEN(INPUT => e.payload:items) item;

-- Running total and rank within group
SELECT
    customer_id,
    order_id,
    amount,
    created_at,
    SUM(amount) OVER (
        PARTITION BY customer_id
        ORDER BY created_at
        ROWS UNBOUNDED PRECEDING
    ) AS lifetime_spend,
    RANK() OVER (
        PARTITION BY DATE_TRUNC('month', created_at)
        ORDER BY amount DESC
    ) AS rank_this_month,
    LAG(amount) OVER (
        PARTITION BY customer_id
        ORDER BY created_at
    ) AS prev_order_amount
FROM orders
WHERE status = 'completed';

-- QUALIFY: filter on window function result (Snowflake extension)
SELECT customer_id, order_id, amount, created_at
FROM orders
QUALIFY ROW_NUMBER() OVER (
    PARTITION BY customer_id
    ORDER BY created_at DESC
) = 1;   -- latest order per customer — no subquery needed

MERGE INTO analytics_prod.silver.customers AS target
USING staging.customer_updates AS source
    ON target.customer_id = source.customer_id
WHEN MATCHED AND source.updated_at > target.updated_at THEN
    UPDATE SET
        name       = source.name,
        email      = source.email,
        updated_at = source.updated_at
WHEN NOT MATCHED THEN
    INSERT (customer_id, name, email, created_at, updated_at)
    VALUES (source.customer_id, source.name, source.email,
            CURRENT_TIMESTAMP(), source.updated_at);

-- Query data at a specific point in the past
SELECT * FROM orders AT (TIMESTAMP => '2026-04-10 09:00:00'::TIMESTAMP_NTZ);

-- Query data before a specific DML statement
SELECT * FROM orders BEFORE (STATEMENT => '01b3d5a8-0000-1234-abcd-1234567890ab');

-- Restore a table to a previous state
CREATE OR REPLACE TABLE orders AS
    SELECT * FROM orders AT (TIMESTAMP => '2026-04-09 00:00:00'::TIMESTAMP_NTZ);

-- Clone table to a historical point (zero-copy — no storage until writes diverge)
CREATE TABLE orders_backup CLONE orders
    AT (TIMESTAMP => '2026-04-15 18:00:00'::TIMESTAMP_NTZ);

-- View table history and associated storage
SELECT *
FROM TABLE(INFORMATION_SCHEMA.TABLE_STORAGE_METRICS(
    TABLE_NAME => 'ORDERS',
    TABLE_SCHEMA => 'SALES'
));

-- Clone entire database for testing
CREATE DATABASE analytics_staging CLONE analytics_prod;

-- Clone at a point in time
CREATE TABLE orders_march CLONE orders
    AT (TIMESTAMP => '2026-03-31 23:59:59'::TIMESTAMP_NTZ);

-- Define a dynamic table — Snowflake manages refresh automatically
CREATE OR REPLACE DYNAMIC TABLE silver.customer_monthly_stats
    TARGET_LAG = '1 hour'        -- max data freshness
    WAREHOUSE = etl_wh
AS
SELECT
    customer_id,
    DATE_TRUNC('month', created_at)    AS month,
    COUNT(*)                            AS order_count,
    SUM(amount)                         AS revenue,
    AVG(amount)                         AS avg_order_value
FROM bronze.orders
WHERE status = 'completed'
GROUP BY 1, 2;

-- Check refresh history
SELECT *
FROM TABLE(INFORMATION_SCHEMA.DYNAMIC_TABLE_REFRESH_HISTORY(
    NAME => 'silver.customer_monthly_stats'
));

from snowflake.snowpark import Session
from snowflake.snowpark.functions import col, sum as sum_, avg, count, lit, when
from snowflake.snowpark.types import IntegerType, FloatType

connection_params = {
    "account":   "my-org.snowflakecomputing.com",
    "user":      "my_service_account",
    "authenticator": "externalbrowser",   # SSO in interactive sessions
    # for service accounts: "private_key_file": "/path/to/key.p8"
    "role":      "ANALYST_ROLE",
    "warehouse": "analytics_wh",
    "database":  "analytics_prod",
    "schema":    "sales",
}

session = Session.builder.configs(connection_params).create()

# DataFrame API — lazy evaluation, pushes SQL to Snowflake
orders_df = session.table("orders")

result = (
    orders_df
    .filter(col("status") == "completed")
    .group_by("customer_id")
    .agg(
        count("*").alias("order_count"),
        sum_("amount").alias("total_spend"),
        avg("amount").alias("avg_order")
    )
    .filter(col("total_spend") > 1000)
    .sort(col("total_spend").desc())
)

result.show()

# Write result back to Snowflake
result.write.mode("overwrite").save_as_table("gold.high_value_customers")

from snowflake.snowpark.functions import udf
from snowflake.snowpark.types import StringType, FloatType
import snowflake.snowpark.functions as F

# Register a scalar UDF — runs per-row inside Snowflake
@udf(name="categorise_amount", replace=True,
     return_type=StringType(), input_types=[FloatType()])
def categorise_amount(amount: float) -> str:
    if amount < 50:
        return "micro"
    elif amount < 500:
        return "standard"
    elif amount < 5000:
        return "large"
    return "enterprise"

# Use the UDF in a DataFrame
orders_df.with_column(
    "category",
    F.call_udf("categorise_amount", col("amount"))
).show()

def transform_orders(session: Session, target_table: str) -> str:
    raw = session.table("bronze.raw_orders")

    clean = (
        raw
        .filter(col("order_id").is_not_null())
        .drop_duplicates(["order_id"])
        .with_column("amount", col("amount").cast(FloatType()))
        .with_column("loaded_at", F.current_timestamp())
    )

    clean.write.mode("overwrite").save_as_table(target_table)
    return f"Loaded {clean.count()} rows into {target_table}"

# Register as a stored procedure
session.sproc.register(
    func=transform_orders,
    name="sp_transform_orders",
    replace=True,
    is_permanent=True,
    stage_location="@my_stage",
    packages=["snowflake-snowpark-python"]
)

# Call it
session.call("sp_transform_orders", "silver.orders")

-- Provider side: create a share
CREATE SHARE product_data_share;

GRANT USAGE ON DATABASE analytics_prod TO SHARE product_data_share;
GRANT USAGE ON SCHEMA analytics_prod.gold TO SHARE product_data_share;
GRANT SELECT ON TABLE analytics_prod.gold.product_catalog TO SHARE product_data_share;
GRANT SELECT ON VIEW analytics_prod.gold.public_pricing TO SHARE product_data_share;

-- Add consumer account
ALTER SHARE product_data_share ADD ACCOUNTS = org_name.consumer_account;

-- Consumer side: create a database from the share
CREATE DATABASE partner_products FROM SHARE provider_org.product_data_share;

-- Query it as a normal database
SELECT * FROM partner_products.gold.product_catalog;

-- After subscribing to a marketplace dataset, it appears as a database
SELECT *
FROM CYBERSYN__COMPANY_RELATIONSHIP_GRAPH.PUBLIC.COMPANY_INDEX
WHERE name ILIKE '%microsoft%'
LIMIT 20;

-- Create a stream on a table
CREATE STREAM orders_stream ON TABLE bronze.raw_orders;

-- The stream captures inserts, updates, deletes with metadata columns:
-- METADATA$ACTION: 'INSERT' or 'DELETE'
-- METADATA$ISUPDATE: TRUE for updates (appear as DELETE + INSERT pair)
-- METADATA$ROW_ID: stable row identifier across DML

-- Consume the stream in a task
SELECT * FROM orders_stream WHERE METADATA$ACTION = 'INSERT';

-- Create a scheduled task
CREATE TASK refresh_silver_orders
    WAREHOUSE = etl_wh
    SCHEDULE = 'USING CRON 0 * * * * UTC'   -- every hour
AS
    CALL sp_transform_orders('silver.orders');

-- Task DAG: child task runs after parent
CREATE TASK refresh_gold_stats
    WAREHOUSE = etl_wh
    AFTER refresh_silver_orders
AS
    CREATE OR REPLACE TABLE gold.monthly_stats AS
    SELECT DATE_TRUNC('month', created_at) AS month,
           SUM(amount) AS revenue
    FROM silver.orders
    GROUP BY 1;

-- Activate tasks
ALTER TASK refresh_gold_stats RESUME;
ALTER TASK refresh_silver_orders RESUME;

-- Row access policy: filter rows based on current role
CREATE ROW ACCESS POLICY tenant_isolation
    AS (tenant_id VARCHAR) RETURNS BOOLEAN ->
    CURRENT_ROLE() IN ('ADMIN_ROLE')
    OR tenant_id = CURRENT_USER();

ALTER TABLE silver.orders ADD ROW ACCESS POLICY tenant_isolation ON (tenant_id);

-- Column masking policy: mask PII for non-privileged roles
CREATE MASKING POLICY email_mask
    AS (val STRING) RETURNS STRING ->
    CASE WHEN CURRENT_ROLE() IN ('PII_ADMIN') THEN val
         ELSE REGEXP_REPLACE(val, '.+@', '*****@')
    END;

ALTER TABLE silver.customers MODIFY COLUMN email
    SET MASKING POLICY email_mask;

-- Credit usage by warehouse (last 30 days)
SELECT warehouse_name,
       SUM(credits_used)          AS total_credits,
       SUM(credits_used) * 3.00   AS estimated_usd  -- approximate Standard rate
FROM snowflake.account_usage.warehouse_metering_history
WHERE start_time >= DATEADD(day, -30, CURRENT_TIMESTAMP())
GROUP BY 1
ORDER BY total_credits DESC;

-- Most expensive queries
SELECT query_text,
       total_elapsed_time / 1000 AS elapsed_sec,
       credits_used_cloud_services,
       bytes_scanned / 1e9        AS gb_scanned,
       partitions_scanned,
       partitions_total,
       (partitions_scanned / NULLIF(partitions_total, 0)) * 100 AS pct_scanned
FROM snowflake.account_usage.query_history
WHERE start_time >= DATEADD(day, -7, CURRENT_TIMESTAMP())
ORDER BY credits_used_cloud_services DESC NULLS LAST
LIMIT 20;