SnowPro Advanced - Data Engineer - Fact Sheet¶

-- Create pipe with auto-ingest
CREATE PIPE my_pipe AUTO_INGEST = TRUE AS
  COPY INTO my_table
  FROM @my_stage/data/
  FILE_FORMAT = (TYPE = 'PARQUET');

-- Check pipe status
SELECT SYSTEM$PIPE_STATUS('my_pipe');

-- View loading history
SELECT * FROM TABLE(INFORMATION_SCHEMA.COPY_HISTORY(
  TABLE_NAME => 'my_table',
  START_TIME => DATEADD('hour', -24, CURRENT_TIMESTAMP())
));

-- Bulk load from stage
COPY INTO my_table
  FROM @my_stage/data/2024/
  FILE_FORMAT = (TYPE = 'CSV' FIELD_DELIMITER = ',' SKIP_HEADER = 1)
  ON_ERROR = 'CONTINUE'
  PURGE = TRUE;

-- Load with transformation
COPY INTO my_table (id, name, created_date)
  FROM (SELECT $1, $2, TO_DATE($3, 'YYYY-MM-DD')
        FROM @my_stage/data/)
  FILE_FORMAT = (TYPE = 'CSV');

-- Validate before loading
COPY INTO my_table
  FROM @my_stage/data/
  VALIDATION_MODE = 'RETURN_ERRORS';

-- Root task (scheduled)
CREATE TASK load_raw
  WAREHOUSE = pipeline_wh
  SCHEDULE = 'USING CRON 0 */1 * * * America/New_York'
AS
  COPY INTO raw_events FROM @event_stage;

-- Child task (runs after parent)
CREATE TASK transform_events
  WAREHOUSE = pipeline_wh
  AFTER load_raw
AS
  INSERT INTO curated_events
  SELECT event_id, event_type, TO_TIMESTAMP(event_ts)
  FROM raw_events_stream;

-- Serverless task
CREATE TASK cleanup_task
  USER_TASK_MANAGED_INITIAL_WAREHOUSE_SIZE = 'XSMALL'
  SCHEDULE = 'USING CRON 0 0 * * * UTC'
AS
  DELETE FROM staging_table WHERE load_date < DATEADD('day', -7, CURRENT_DATE());

-- Resume tasks (root task last)
ALTER TASK transform_events RESUME;
ALTER TASK load_raw RESUME;

-- Create standard stream
CREATE STREAM orders_stream ON TABLE orders;

-- Create append-only stream (only inserts, better for append workloads)
CREATE STREAM events_stream ON TABLE events APPEND_ONLY = TRUE;

-- Consume stream in transformation
INSERT INTO order_summary
  SELECT order_date, SUM(amount) AS daily_total
  FROM orders_stream
  WHERE METADATA$ACTION = 'INSERT'
  GROUP BY order_date;

-- Check if stream has data
SELECT SYSTEM$STREAM_HAS_DATA('orders_stream');

-- Task triggered by stream data
CREATE TASK process_orders
  WAREHOUSE = pipeline_wh
  SCHEDULE = '1 MINUTE'
  WHEN SYSTEM$STREAM_HAS_DATA('orders_stream')
AS
  MERGE INTO order_summary t
  USING (SELECT * FROM orders_stream) s
  ON t.order_id = s.order_id
  WHEN MATCHED AND s.METADATA$ACTION = 'DELETE' THEN DELETE
  WHEN MATCHED THEN UPDATE SET t.amount = s.amount
  WHEN NOT MATCHED AND s.METADATA$ACTION = 'INSERT' THEN INSERT VALUES (s.order_id, s.amount);

-- Create dynamic table with target lag
CREATE DYNAMIC TABLE daily_summary
  TARGET_LAG = '1 hour'
  WAREHOUSE = pipeline_wh
AS
  SELECT DATE_TRUNC('day', order_date) AS summary_date,
         region,
         COUNT(*) AS order_count,
         SUM(amount) AS total_revenue
  FROM orders
  GROUP BY 1, 2;

-- Chain dynamic tables
CREATE DYNAMIC TABLE regional_rollup
  TARGET_LAG = '2 hours'
  WAREHOUSE = pipeline_wh
AS
  SELECT summary_date, region,
         total_revenue,
         total_revenue / SUM(total_revenue) OVER (PARTITION BY summary_date) AS revenue_pct
  FROM daily_summary;

-- Monitor refresh history
SELECT * FROM TABLE(INFORMATION_SCHEMA.DYNAMIC_TABLE_REFRESH_HISTORY(
  NAME => 'daily_summary'
));

-- Task execution history
SELECT * FROM TABLE(INFORMATION_SCHEMA.TASK_HISTORY(
  SCHEDULED_TIME_RANGE_START => DATEADD('hour', -24, CURRENT_TIMESTAMP()),
  RESULT_LIMIT => 100
));

-- Pipe usage history
SELECT * FROM TABLE(INFORMATION_SCHEMA.PIPE_USAGE_HISTORY(
  DATE_RANGE_START => DATEADD('day', -7, CURRENT_TIMESTAMP())
));

-- Copy history
SELECT * FROM TABLE(INFORMATION_SCHEMA.COPY_HISTORY(
  TABLE_NAME => 'my_table',
  START_TIME => DATEADD('day', -1, CURRENT_TIMESTAMP())
));

-- Dynamic table refresh history
SELECT * FROM TABLE(INFORMATION_SCHEMA.DYNAMIC_TABLE_REFRESH_HISTORY());