Snowflake Tasks for Document Pipelines: Triggers, Compute and Failures

1.Scheduled tasks versus stream-triggered tasks

A task runs a SQL statement, which can include a stored procedure call. It runs either on a fixed schedule or when an event fires. A scheduled task uses the SCHEDULE parameter, which takes either an interval such as '60 MINUTES' or a USING CRON expression with a time zone. Snowflake runs only one instance of a scheduled task at a time. "If a task is still running when the next scheduled run time occurs, then that scheduled time is skipped."

Documents tend to arrive at unpredictable times, which is the case triggered tasks are built for. Instead of a schedule, the task has a WHEN clause that calls SYSTEM$STREAM_HAS_DATA. The documentation says this "eliminates frequent polling of the source when new data arrival is unpredictable" and also cuts latency, because data is processed as soon as it arrives.

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

You can also combine the two, so that a scheduled task checks the stream once an hour and runs only when there is something to process. Whichever form you choose, a new task does nothing until you resume it. A task is created suspended. ALTER TASK … RESUME lets it run continuously, and EXECUTE TASK runs it once, which is useful for testing.

CREATE TASK triggered_task_stream
  SCHEDULE = '1 HOUR'
  WHEN  ? ('orders_stream')
  AS SELECT 1;

Sources12

2.Triggering on new files: the directory-table refresh

Directory tables appear on the list of supported sources for triggered tasks, with one condition. The directory table must be refreshed before the stream sees a new file. You can set the directory table to auto-refresh, or refresh it yourself with ALTER STAGE name REFRESH. Streams on external tables and hybrid tables are not supported for triggered tasks. The documentation includes an example task that fires on a directory-table stream and records file metadata, including the file URL and the stream's action column.

CREATE TASK triggered_task_directory_table
  WAREHOUSE = my_warehouse
  WHEN SYSTEM$STREAM_HAS_DATA('my_directory_table_stream')
  AS
    INSERT INTO tasks_runs
    SELECT 'trigger_t_internal_stage', relative_path, size,
            last_modified, file_url, etag, metadata$action
    FROM my_directory_table_stream;

Once resumed, the task checks the stream. If there are changes it runs. If not, it skips the run without using compute. If more data arrives while a run is in progress, the next run waits until the current one finishes. By default, triggered tasks run at most every 30 seconds. Setting USER_TASK_MINIMUM_TRIGGER_INTERVAL_IN_SECONDS can lower that to 10 seconds. If a task hasn't run for 12 hours, Snowflake schedules a health check to keep the stream from going stale. Even so, the task body still has to consume the stream before data retention expires.

Sources3

3.Serverless or warehouse compute

Every task needs compute, and there are two models. With serverless tasks, Snowflake predicts the resources each run needs from recent runs and assigns them for you. You leave out the WAREHOUSE parameter, and the role needs the global EXECUTE MANAGED TASK privilege. With user-managed tasks, you name a warehouse and size it yourself.

You can put bounds on serverless sizing with SERVERLESS_TASK_MIN_STATEMENT_SIZE (default XSMALL) and SERVERLESS_TASK_MAX_STATEMENT_SIZE (default XXLARGE). Triggered tasks have an extra requirement. A triggered task has no schedule, so a serverless triggered task must declare TARGET_COMPLETION_INTERVAL. Snowflake sizes resources to finish within that interval.

CREATE TASK my_triggered_task
   ? ='15 MINUTES'
  WHEN SYSTEM$STREAM_HAS_DATA('my_order_stream')
  AS
    INSERT INTO customer_activity
    SELECT customer_id, order_total, order_date, 'order'
    FROM my_order_stream;

Sources1

4.Failures, retries and multi-step task graphs

A pipeline that keeps failing still costs credits. SUSPEND_TASK_AFTER_NUM_FAILURES suspends a task once it has failed or timed out a set number of times in a row. You can set it on the task, or at account, database or schema level, and a lower-level setting overrides a higher one. TASK_AUTO_RETRY_ATTEMPTS retries a task that ends in a FAILED state. Automatic retry is off by default.

When processing has several stages, such as one step that writes raw output and later steps that build on it, you can chain tasks into a task graph (a DAG). The root task defines when the graph runs, either on a schedule or by trigger. Child tasks are created with CREATE TASK … AFTER. Siblings run in parallel, and a task with several parents waits for all of them. An optional finalizer, attached with FINALIZE = <root>, runs after everything else for cleanup or notifications. Every task in a graph must have the same owner and live in the same database and schema.

CREATE OR REPLACE TASK task_root
  SCHEDULE = '1 MINUTE'
  TASK_AUTO_RETRY_ATTEMPTS = 2   --  Failed task graph retries up to 2 times
  SUSPEND_TASK_AFTER_NUM_FAILURES = 3   --  Task graph suspends after 3 consecutive failures
  AS SELECT 1;

By default, one failed child task means the whole graph run has failed, and the graph is suspended after 10 consecutive failures. You can retry from the point of failure with EXECUTE TASK … RETRY LAST. To change any task in the graph, suspend the root first. A run already in progress will finish.

Sources14

Exam traps

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

1. 1.A triggered task on a directory-table stream fires as soon as a document is uploaded to the stage.Why is that wrong?

   The directory table has to be refreshed first, either by auto-refresh or by ALTER STAGE … REFRESH, before the stream and task can see the new file.

   Covered in Triggering on new files: the directory-table refresh

2. 2.A serverless triggered task only needs a WHEN clause; omitting WAREHOUSE is enough to make it serverless.Why is that wrong?

   A serverless triggered task also has to set TARGET_COMPLETION_INTERVAL, because it has no schedule for Snowflake to size against.

   Covered in Serverless or warehouse compute

3. 3.A task begins running on its schedule or trigger as soon as CREATE TASK succeeds.Why is that wrong?

   New tasks start suspended. You need ALTER TASK … RESUME to start them, or EXECUTE TASK for a single run.

   Covered in Scheduled tasks versus stream-triggered tasks

Practise it for real

Build a triggered task that records every file landing on a stage, using a stream on the stage's directory table.

1. 1.Create an internal stage named my_test_stage with DIRECTORY = ( ENABLE = TRUE ) and SNOWFLAKE_SSE encryption, then create a tasks_runs table whose columns match the SELECT list in the triggered-task example.

   Why: The stream needs a directory table to track, and the task needs a target table to write to.

   You should see: Both objects exist. LIST @my_test_stage returns no files.

2. 2.Run CREATE STREAM my_directory_table_stream ON STAGE my_test_stage;

   Why: The stream sets its offset now and reports file-level changes made after this point.

   You should see: SELECT * FROM my_directory_table_stream returns zero rows.

3. 3.Create triggered_task_directory_table exactly as in the triggered-task example, then run ALTER TASK triggered_task_directory_table RESUME.

   Why: Tasks are created suspended, so the trigger isn't watched until you resume.

   You should see: SHOW TASKS lists the task as started, with SCHEDULE shown as NULL.

4. 4.Run COPY INTO @my_test_stage/my_test_file FROM (SELECT 100) OVERWRITE=TRUE to write a file to the stage.

   Why: This lands a new file, but the directory table doesn't know about it yet.

   You should see: The task doesn't run yet, because the directory table hasn't been refreshed.

5. 5.Run ALTER STAGE my_test_stage REFRESH, wait at least 30 seconds, then query tasks_runs.

   Why: The refresh adds the file to the directory table. The stream picks it up and the task consumes it.

   You should see: tasks_runs has a row for my_test_file with metadata$action = INSERT, and the stream is empty again.