snowflake-core-workflow-b

name	snowflake-core-workflow-b
description	Execute Snowflake data transformation with streams, tasks, and dynamic tables. Use when building ELT pipelines, scheduling transformations, or implementing change data capture with Snowflake streams. Trigger with phrases like "snowflake transform", "snowflake ELT", "snowflake stream", "snowflake task", "snowflake pipeline", "snowflake dynamic table", "snowflake CDC".
allowed-tools	Read, Write, Edit, Bash(npm:*), Grep
version	1.5.0
license	MIT
author	Jeremy Longshore <[e-mail]>
tags	["saas","data-warehouse","analytics","snowflake"]
compatibility	Designed for Claude Code

Snowflake Core Workflow B — Data Transformation

Overview

Build ELT pipelines using streams (change data capture), tasks (scheduling), and dynamic tables (declarative transforms).

Prerequisites

• Data loaded into Snowflake (via snowflake-core-workflow-a)
• Understanding of ELT vs ETL patterns
• Role with CREATE TASK, CREATE STREAM privileges

Instructions

Step 1: Create a Stream for Change Data Capture

-- Track changes on the raw orders table
CREATE OR REPLACE STREAM orders_stream ON TABLE raw_orders
  APPEND_ONLY = FALSE;

-- Append-only stream (lighter weight, inserts only)
CREATE OR REPLACE STREAM events_stream ON TABLE raw_events
  APPEND_ONLY = TRUE;

-- Check what's changed since last consumption
SELECT * FROM orders_stream;
-- METADATA$ACTION = 'INSERT' | 'DELETE'
-- METADATA$ISUPDATE = TRUE if row is part of an UPDATE
-- METADATA$ROW_ID = unique row identifier

Step 2: Create a Task to Process Stream Data

-- Transform task runs when stream has data
CREATE OR REPLACE TASK transform_orders
  WAREHOUSE = TRANSFORM_WH
  SCHEDULE = '5 MINUTE'
  WHEN SYSTEM$STREAM_HAS_DATA('orders_stream')
AS
  MERGE INTO dim_orders AS target
   (
    
      order_id,
      customer_id,
      amount::(,)  amount,
      order_date::TIMESTAMP_NTZ  order_date,
      
         amount    
         amount    
         
        order_tier,
      ()  processed_at
     orders_stream
     METADATA$ACTION  
  )  source
   target.order_id  source.order_id
   MATCHED   
    target.amount  source.amount,
    target.order_tier  source.order_tier,
    target.processed_at  source.processed_at
    MATCHED  
    (order_id, customer_id, amount, order_date, order_tier, processed_at)
  
    (source.order_id, source.customer_id, source.amount,
     source.order_date, source.order_tier, source.processed_at);


 TASK transform_orders RESUME;