Streams, tasks, dynamic tables and programmatic pipelines

1.Streams: change data capture with an offset

After data lands, a continuous pipeline has to process only what has changed. A stream records DML changes (inserts, including COPY INTO, plus updates and deletes) on a source object. You can then query those changes as a change table. Streams can be created on standard and shared tables, views (including secure views), directory tables, dynamic tables, Iceberg tables (with limitations), event tables and external tables.

A stream doesn't contain any table data. It stores only an offset, which works like a bookmark between two table versions. It then returns change records from the source object's version history. When you create the first stream on a table, Snowflake adds hidden change-tracking columns to that table. Views are an exception: for a stream on a view, you must enable change tracking explicitly on the view and on its underlying tables.

The offset advances only when a DML transaction consumes the stream. CTAS and COPY INTO location count as DML here. Running a plain SELECT on a stream, even inside an explicit transaction, leaves the offset where it is. If several statements must see the same change records, wrap them in BEGIN … COMMIT. This locks the stream, and because streams use repeatable read isolation, every statement in the transaction sees the same set of changes.

Sources1

2.Tasks: schedules and stream triggers

A task runs a SQL command or a stored procedure on a fixed schedule or when an event happens. Stored procedures can be written in JavaScript, Python, Java, Scala or Snowflake Scripting. A schedule can be an interval such as '10 SECONDS', or a CRON expression with a time zone. Only one instance of a scheduled task runs at a time. If a run is still going when the next one is due, that next run is skipped.

A triggered task uses WHEN SYSTEM$STREAM_HAS_DATA(...) and runs when the stream has new data. When data arrives unpredictably, this avoids frequent polling and processes new data sooner. You can also add a SCHEDULE to a triggered task, so the stream check runs at that interval. Groups of tasks can be chained into task graphs that run steps in parallel or in sequence.

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;

A new task starts in the suspended state. Use EXECUTE TASK to test it once, and ALTER TASK … RESUME to start its schedule or event detection.

For multi-step pipelines, build a task graph with parent-child dependencies, fan-out and fan-in patterns, and finalizer tasks. When you create a task, you also define what happens when it fails. For example, SUSPEND_TASK_AFTER_NUM_FAILURES = 3 suspends a task after three failures. To diagnose problems such as an auto-suspended task, or a task that works interactively but fails under the scheduler, inspect TASK_HISTORY and the owner role's privileges.

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

Sources23

3.Serverless or user-managed task compute

Every task needs compute. If you leave out the WAREHOUSE parameter, the task is serverless. Snowflake predicts the compute it needs from recent runs, within SERVERLESS_TASK_MIN_STATEMENT_SIZE and SERVERLESS_TASK_MAX_STATEMENT_SIZE. The defaults are XSMALL and XXLARGE. If you include WAREHOUSE, the task is user-managed, and you control the compute size directly. In both cases, the role that runs the task needs the global EXECUTE MANAGED TASK privilege.

TARGET_COMPLETION_INTERVAL tells Snowflake to scale a serverless task so it finishes sooner. It is required for serverless triggered tasks. When a task has already reached its maximum size, Snowflake ignores the interval.

CREATE TASK SCHEDULED_T2
  SCHEDULE='30 SECONDS'
  SERVERLESS_TASK_MIN_STATEMENT_SIZE='SMALL'
  SERVERLESS_TASK_MAX_STATEMENT_SIZE='LARGE'
  AS SELECT 1;

Sources2

4.Dynamic tables and materialized views

Streams and tasks require you to write the orchestration yourself. A dynamic table is the declarative alternative. You write a SELECT and set a TARGET_LAG, and Snowflake works out the dependency graph and keeps the results fresh. Each refresh is applied atomically. When dynamic tables read from one another, Snowflake refreshes them in dependency order against a consistent snapshot.

CREATE OR ALTER DYNAMIC TABLE dt_orders
    TARGET_LAG = '10 minutes'
    WAREHOUSE = transform_wh
    REFRESH_MODE = INCREMENTAL
AS
    SELECT
        order_id,
        customer_id,
        order_date,
        TRIM(UPPER(product_name)) AS product_name,
        quantity,
        unit_price,
        quantity * unit_price AS line_total,
        order_status
    FROM raw_orders
    WHERE order_status != 'returned';

Target lag is a goal, not a guarantee: the actual lag can exceed it if a refresh runs long. Setting TARGET_LAG = DOWNSTREAM on an intermediate table makes it refresh only when the tables that depend on it need fresh data. The refresh modes are INCREMENTAL, FULL, AUTO (chosen at creation), ADAPTIVE and CUSTOM_INCREMENTAL. Costs come from warehouse compute, cloud services and storage.

Dynamic tables don't fit workloads that need data fresher than 60 seconds, the minimum target lag, or that need strictly guaranteed refresh timing. They also can't use stored procedures or external functions in their definition. For those cases, use streams with tasks.

A materialized view solves a different problem. It precomputes a query result to speed up frequent or expensive aggregation, projection and selection on large data sets. In other words, it makes reads faster, whereas a dynamic table is a step in a transformation pipeline.

Materialized views have trade-offs and limits worth knowing:

- Always current. A background service maintains the view automatically. If a query runs before the view is up to date, Snowflake updates it or reads the newer data from the base table. A dynamic table has a minimum lag of 1 minute, so use a materialized view when you need always-current data with no lag. - Cost. Keeping the view up to date uses compute, and the stored results use storage. A materialized view is worth it only when the saved work outweighs those costs. - Single base table. Materialized views accelerate repeated queries against a single base table, with no joins. Dynamic tables are the choice for joins, aggregations and multi-step pipelines. - Transparent use. The optimizer rewrites queries to use a materialized view automatically. That doesn't happen with dynamic tables. - When to create one. Create one when the results don't change often, are used often, and the query consumes a lot of resources. Otherwise a regular view is the better fit.

In short, use a materialized view for read performance on one table, and a dynamic table to build a pipeline.

Sources4567

5.UDFs, Snowflake Scripting and Notebooks

Pipeline logic that you reuse can live in a UDF. CREATE FUNCTION defines a function that returns either scalar or tabular results. Its handler is written in a supported language. Depending on the language, the handler code is either written inline in the statement or kept on a stage. CREATE OR ALTER FUNCTION creates the function if it doesn't exist, or alters it if it does.

For procedural steps, such as branching, variables and error handling, use Snowflake Scripting. You write the code in a block, which tasks can run inside stored procedures. Only BEGIN and END are required. DECLARE and EXCEPTION are optional. A block's BEGIN isn't the same as the BEGIN that starts a transaction, so Snowflake recommends starting transactions with BEGIN TRANSACTION. This matters when a block also has to consume a stream inside a transaction.

DECLARE
  -- (variable declarations, cursor declarations, etc.) ...
BEGIN
  -- (Snowflake Scripting and SQL statements) ...
EXCEPTION
  -- (statements for handling exceptions) ...
END;

The exam guide also lists using Notebooks to run pipelines of stored procedures for ingestion. None of the sources for this lesson covers Notebooks, so this lesson can't describe how they work. Check the Notebooks documentation directly for that bullet.

Sources892

6.Driving pipelines through the SQL API

External orchestrators can run pipeline SQL through the Snowflake SQL API, a REST API for running queries and most DDL and DML. It has three endpoints under /api/v2/statements/: one submits statements, one checks a statement's status by its statement handle, and one cancels a statement. You submit with a POST that sends the statement and, optionally, a timeout, bind variables, and the warehouse, database, schema and role. If you leave out the context fields, the API uses the user's defaults. These values are case-sensitive and must match what SHOW returns.

The API authenticates with OAuth or a key pair. In the documented curl example, the request carries a JWT that you generated in an Authorization: Bearer header.

{
  "statement": "select * from T where c1=?",
  "timeout": 60,
  "database": "TESTDB",
  "schema": "TESTSCHEMA",
  "warehouse": "TESTWH",
  "role": "TESTROLE",
  "bindings": {
    "1": {
      "type": "FIXED",
      "value": "123"
    }
  }
}

Your client must handle asynchronous responses. If async=true isn't set, the API returns results only when the statement finishes within 45 seconds. Otherwise it returns a statement handle and HTTP 202, and you poll with GET. A 408 means the statement exceeded its timeout and was cancelled. A 422 means it failed.

The API has limits that matter for pipelines. PUT and GET aren't supported, so you can't use it to upload files to a stage. BEGIN, COMMIT, ROLLBACK, USE, ALTER SESSION and temporary-object creation work only inside a multi-statement request. AUTOCOMMIT must be TRUE at the statement level. Some Python and Java/Scala stored procedures that return Arrow result sets can fail.

Sources10111213

Exam traps

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

1. 1.Selecting from a stream marks those changes as consumed.Why is that wrong?

   Only a committed DML transaction that uses the stream advances its offset. Plain queries can read the same changes as many times as needed.

   Covered in Streams: change data capture with an offset

2. 2.A serverless triggered task needs only the WHEN condition; Snowflake picks the timing.Why is that wrong?

   Serverless triggered tasks must set TARGET_COMPLETION_INTERVAL.

   Covered in Serverless or user-managed task compute

3. 3.Setting TARGET_LAG guarantees the dynamic table is never staler than that value.Why is that wrong?

   Target lag is a goal. Actual lag can be longer when refreshes run long, and the minimum target is 60 seconds.

   Covered in Dynamic tables and materialized views

Practise it for real

Turn a polling job into a stream-triggered task and bring it online safely.

1. 1.Create a user-managed triggered task: CREATE TASK triggered_task_stream WAREHOUSE = transform_wh WHEN SYSTEM$STREAM_HAS_DATA('orders_stream') AS INSERT INTO completed_promotions SELECT ... FROM orders_stream; (transform_wh is a warehouse you own.)

   Why: With a WHEN condition, the task runs only when the stream has new data, so it doesn't poll for nothing. Including WAREHOUSE makes it user-managed, so no TARGET_COMPLETION_INTERVAL is needed.

   You should see: The task exists in a suspended state, because every new task starts suspended.

2. 2.Run EXECUTE TASK triggered_task_stream;

   Why: Manually testing one run is the documented step before you resume a task.

   You should see: A single run that inserts any pending stream rows into completed_promotions and advances the stream offset.

3. 3.Run ALTER TASK triggered_task_stream RESUME;

   Why: Resuming lets the task detect stream events continuously.

   You should see: The task now runs whenever orders_stream has new data.

4. 4.Review the task's costs and history, then refine it with ALTER TASK.

   Why: Monitoring costs and then refining with ALTER TASK are the final steps in the documented task workflow.

   You should see: Settings adjusted based on observed runs.