Snowflake Data Engineering

Snowflake Data Engineering

This skill covers SQL conventions, pipeline architecture (Dynamic Tables, Streams, Tasks, Snowpipe), performance tuning, and cost/access management on Snowflake.

Workflow for Building a Snowflake Pipeline

1. Land raw data — Use Snowpipe (AUTO_INGEST = TRUE) for continuous file loads from an external stage, or Snowpipe Streaming for low-latency row-level ingestion via SDK.
2. Choose a transformation approach — Prefer Dynamic Tables for declarative, most pipelines; fall back to Streams + Tasks only when you need procedural logic or stored-procedure calls.
3. Model semi-structured data — Land raw JSON/Avro/Parquet as VARIANT, then flatten into typed relational columns as early as practical.
4. Chain pipeline stages — Build Dynamic Tables on top of each other (or Streams feeding Tasks) so each stage narrows scope from raw to cleaned to aggregated.
5. Tune for performance — Add clustering keys or Search Optimization only where query patterns justify them; tag queries for cost attribution.
6. Set access controls — Apply least-privilege RBAC with functional roles (loader, transformer, analyst) and masking/row-access policies for sensitive data.
7. Monitor cost and freshness — Track WAREHOUSE_METERING_HISTORY and QUERY_HISTORY, set Resource Monitors, and validate TARGET_LAG matches actual freshness requirements.

SQL and Semi-Structured Data

• Use VARIANT, OBJECT, and ARRAY types for JSON, Avro, Parquet, and ORC data.
• Access nested fields with colon notation and cast explicitly: src:customer.name::STRING, src:price::NUMBER(10,2), src:created_at::TIMESTAMP_NTZ.
• Flatten arrays with LATERAL FLATTEN:

SELECT f.value:name::STRING AS name
FROM my_table, LATERAL FLATTEN(input => src:items) f;

• Flatten semi-structured data into relational columns whenever it contains dates, numbers stored as strings, or arrays — keeping data inside VARIANT prevents Snowflake's automatic subcolumnarization from paying off.
• Avoid mixing types within the same VARIANT field for the same reason.
• Remember that a JSON null is stored as the string "null", distinct from a SQL NULL. Use STRIP_NULL_VALUES => TRUE on load when you want them treated the same.

SQL coding standards

• Use snake_case for all identifiers; avoid quoted identifiers.
• Prefer CTEs over deeply nested subqueries for readability.
• Use CREATE OR REPLACE for idempotent DDL.
• Use COPY INTO for bulk loading, never row-by-row INSERT.
• Use MERGE for upserts:

MERGE INTO target t USING source s ON t.id = s.id
WHEN MATCHED THEN UPDATE SET t.name = s.name
WHEN NOT MATCHED THEN INSERT (id, name) VALUES (s.id, s.name);

• In stored procedures, prefix session/local variables with : when referencing them inside SQL statements:

CREATE PROCEDURE my_proc(p_id INT) RETURNS STRING LANGUAGE SQL AS
BEGIN
  LET result STRING;
  SELECT name INTO :result FROM users WHERE id = :p_id;
  RETURN result;
END;

Performance Optimization

• Add cluster keys only for very large tables (multi-TB) on columns frequently used in WHERE/JOIN/GROUP BY:

ALTER TABLE large_events CLUSTER BY (event_date, region);

• Use the Search Optimization Service for point lookups on high-cardinality columns or substring/regex search:

ALTER TABLE logs ADD SEARCH OPTIMIZATION ON EQUALITY(sender_ip), SUBSTRING(error_message);

• Use Materialized Views to pre-compute expensive single-table aggregations.
• Reuse prior results with RESULT_SCAN(LAST_QUERY_ID()) instead of re-running an identical query.
• Tag queries for cost attribution: ALTER SESSION SET QUERY_TAG = 'etl_daily_load';
• Never run SELECT * on wide tables — it defeats columnar pruning benefits.

Data Pipelines

Choose the right primitive:

Dynamic Tables

CREATE OR REPLACE DYNAMIC TABLE cleaned_events
  TARGET_LAG = '5 minutes'
  WAREHOUSE = transform_wh
  AS
  SELECT event_id, event_type, user_id, event_data:page::STRING AS page, event_timestamp
  FROM raw_events
  WHERE event_type IS NOT NULL;

-- Chain for multi-step pipelines
CREATE OR REPLACE DYNAMIC TABLE user_sessions
  TARGET_LAG = '10 minutes'
  WAREHOUSE = transform_wh
  AS
  SELECT user_id, MIN(event_timestamp) AS session_start, MAX(event_timestamp) AS session_end,
         COUNT(*) AS event_count
  FROM cleaned_events GROUP BY user_id;

TARGET_LAG sets the freshness target. REFRESH_MODE can be AUTO, FULL, or INCREMENTAL. Manage lifecycle with ALTER DYNAMIC TABLE ... SET TARGET_LAG / REFRESH / SUSPEND / RESUME.

Streams (CDC)

CREATE OR REPLACE STREAM raw_events_stream ON TABLE raw_events;

Streams add METADATA$ACTION, METADATA$ISUPDATE, and METADATA$ROW_ID columns. Set APPEND_ONLY = TRUE for insert-only sources to lower overhead.

Tasks (scheduled/triggered)

CREATE OR REPLACE TASK process_events
  WAREHOUSE = transform_wh
  SCHEDULE = 'USING CRON 0 */1 * * * America/Los_Angeles'
  WHEN SYSTEM$STREAM_HAS_DATA('raw_events_stream')
  AS
  INSERT INTO cleaned_events
  SELECT event_id, event_type, user_id, event_timestamp
  FROM raw_events_stream WHERE event_type IS NOT NULL;

Build Task DAGs with CREATE TASK child_task ... AFTER parent_task .... Tasks are created SUSPENDED by default — remember ALTER TASK ... RESUME or nothing will run.