Snowflake Pipeline Automation: Data Quality, Tasks and Task Graphs

1.Cleanse and conform: measuring quality with data metric functions

Before you cleanse data, you have to know what is wrong with it. In Snowflake that starts with data metric functions (DMFs). Snowflake ships system DMFs in the CORE schema of the shared SNOWFLAKE database. Each one measures a single quality attribute: NULL_COUNT counts NULLs, DUPLICATE_COUNT counts duplicated values (NULLs included), UNTRIMMED_STRING_COUNT finds values with leading or trailing whitespace, CASE_FORMAT_VIOLATION_COUNT finds strings with inconsistent casing, and ACCEPTED_VALUES determines whether values in a column match a Boolean expression you supply. Several of these map directly onto conforming work, such as trimming, standardising case, deduplicating and enforcing allowed values.

You can attach DMFs when you create the object, so monitoring starts right away. Each binding can carry an EXPECTATION. The left side of the expectation must be the keyword VALUE, so VALUE = 0 means "this metric should be zero". All bindings attach atomically: if one is invalid, the whole CREATE fails.

CREATE OR REPLACE TABLE orders (
  order_id    NUMBER,
  customer_id NUMBER
)
WITH DATA METRIC FUNCTION (
  SNOWFLAKE.CORE.NULL_COUNT
    ON (customer_id)
    EXPECTATION no_null_customers ( VALUE = 0 ),
  SNOWFLAKE.CORE.DUPLICATE_COUNT
    ON (order_id)
    EXPECTATION no_duplicate_orders ( VALUE = 0 )
);

The object parameter DATA_METRIC_SCHEDULE controls when DMFs run. You can set it to a number of minutes, a CRON expression, or 'TRIGGER_ON_CHANGES' to run after DML changes the table. The trigger approach is only available for certain kinds of tables, and reclustering a table does not trigger a DMF. A table has only one schedule, which all of its DMFs share.

Once a DMF has flagged a problem, the fix is ordinary SQL. The dynamic table examples in the documentation show the common moves. Standardising text: TRIM(UPPER(product_name)) AS product_name removes stray whitespace and fixes casing in one expression. Deduplicating: for append-only change data capture (CDC) records, QUALIFY ROW_NUMBER keeps only the latest row per business key. Enriching: join the cleansed fact data to a dimension or reference table, for example the join of raw orders to dim_customers that adds region and segment columns.

CREATE OR REPLACE DYNAMIC TABLE dt_customers_current
    TARGET_LAG = '10 minutes'
    WAREHOUSE = transform_wh
    REFRESH_MODE = INCREMENTAL
AS
    SELECT * EXCLUDE (raw_metadata_col)
    FROM raw_customers_cdc
    QUALIFY ROW_NUMBER() OVER (
        PARTITION BY customer_id
        ORDER BY updated_at DESC
    ) = 1;

Sources1234

2.Automating steps with tasks: compute and scheduling

A task runs a SQL statement or a stored procedure. When you create one with CREATE TASK, you decide two things: what compute it uses and when it runs. For compute, a serverless task leaves out the WAREHOUSE parameter. Snowflake then sizes the compute from recent runs, up to the equivalent of an XXLARGE warehouse, and you can bound it with SERVERLESS_TASK_MIN_STATEMENT_SIZE and SERVERLESS_TASK_MAX_STATEMENT_SIZE. A user-managed task names a warehouse instead.

For timing, the SCHEDULE parameter takes either an interval such as '60 MINUTES' or a CRON expression with a time zone. Snowflake runs only one instance of a scheduled task at a time. If a run is still going when the next scheduled time arrives, that scheduled time is skipped.

CREATE TASK SCHEDULED_T1
  WAREHOUSE='COMPUTE_WH'
  SCHEDULE='60 MINUTES'
  AS SELECT 1;

CREATE TASK task_sunday_3_07_am_pacific_time_zone
  SCHEDULE='USING CRON 7 3 * * SUN America/Los_Angeles'  -- Use a random minute such as 7
AS SELECT 1;

When new data arrives unpredictably, a triggered task is a better fit. It runs whenever a stream has data, so you don't need to poll the source on a fixed interval. A serverless triggered task requires a target completion interval. You can also combine the two: add a SCHEDULE and a WHEN condition, and the task checks the stream on that schedule. The WHEN condition is evaluated before the task does any work. If it returns FALSE, the run is recorded as SKIPPED in the task history and does not resume the warehouse (for a task on customer-managed compute) or execute the task's SQL. That is why a WHEN check on a schedule is cheaper than a task that always runs and then looks for data.

Dynamic tables use a different, declarative model. You supply a SELECT query and a TARGET_LAG, and Snowflake tracks dependencies and refreshes the table. When dynamic tables read from each other, Snowflake infers the dependency graph from the queries, so you don't declare dependencies or script the execution order. TARGET_LAG sets how stale the data is allowed to get, not how often a refresh runs. On intermediate tables you can set TARGET_LAG = DOWNSTREAM so they refresh only when a downstream table needs fresh data.

CREATE TASK triggered_task_stream
  WHEN  ? ('orders_stream')
  AS
    INSERT INTO completed_promotions
    SELECT order_id, order_total, order_time, promotion_id
    FROM orders_stream;

The documented workflow is: define the task with CREATE TASK, test it manually with EXECUTE TASK, let it run continuously with ALTER TASK ... RESUME, then monitor its costs and refine it with ALTER TASK.

Sources564

3.Chaining steps into a task graph

A real pipeline has several steps, for example landing data, cleansing it and then publishing it. A task graph (a DAG) links tasks together. The root task holds the schedule or trigger. Each child task names its parents with AFTER. Children that share a parent run in parallel. A task with several parents waits until all of them have completed successfully, although it may also run when some parent tasks are skipped. A graph can hold at most 1000 tasks, and a single task can have at most 100 parents and 100 children.

CREATE TASK task_root
  SCHEDULE = '1 MINUTE'
  AS SELECT 1;

CREATE TASK task_a
  AFTER task_root
  AS SELECT 1;

CREATE TASK task_b
  AFTER task_root
  AS SELECT 1;

CREATE TASK task_c
  AFTER task_a, task_b
  AS SELECT 1;

You can add one optional finalizer task with CREATE TASK ... FINALIZE = <root>. It runs after every other task has completed or failed, which makes it the place for cleanup work. A root task can have only one finalizer, and a finalizer cannot have children. Every task in a graph must have the same owner and live in the same database and schema. Before the graph can run, resume each child task and the finalizer, then resume the root. Calling SYSTEM$TASK_DEPENDENTS_ENABLE on the root resumes them all in one step.

By default only one instance of a graph runs at a time. If a full run takes longer than the root's interval, at least one run is skipped. You control this with OVERLAP_POLICY on the root task. NO_OVERLAP is the default. ALLOW_CHILD_OVERLAP starts a new graph instance while children are still running, but root tasks never overlap. ALLOW_ALL_OVERLAP lets entire graphs run concurrently. Only allow overlap if concurrent runs cannot write incorrect or duplicate data.

Sources7

Exam traps

Each one states something that sounds right. Open it to see what is actually true.

1. 1.A task with a SCHEDULE starts running as soon as CREATE TASK completes.Why is that wrong?

   New tasks are suspended. You need ALTER TASK ... RESUME to start the schedule, or EXECUTE TASK for a single run.

   Covered in Automating steps with tasks: compute and scheduling

2. 2.A dynamic table with TARGET_LAG = '10 minutes' refreshes every 10 minutes, the same way a task SCHEDULE works.Why is that wrong?

   Target lag sets the maximum staleness Snowflake aims for. It is not a fixed refresh interval.

   Covered in Automating steps with tasks: compute and scheduling

3. 3.If a scheduled task overruns its interval, Snowflake queues the next run and starts it when the current run finishes.Why is that wrong?

   Only one instance of a scheduled task runs at a time. The missed scheduled time is skipped, not queued.

   Covered in Automating steps with tasks: compute and scheduling