Tag: snowflake

The 2 a.m. page said the pipeline “succeeded.” The dashboard was green. And the finance team was still staring at yesterday’s numbers, because one task in a forty-task DAG had quietly processed the wrong micro-batch window and nobody could prove when, or why, without SSH-ing into a worker and grepping logs by hand. That’s the gap between “DAG success/failure notifications” and actual observability: a green checkmark tells you the code didn’t throw, not that the right data moved in the right window at the right time.

The fix isn’t a fancier alerting tool. It’s an audit table — a row written to Snowflake at the start and end of every single task, carrying the execution context Airflow already knows: which logical date this run is for, which try number, when the task actually started and finished, how long it took, and what it touched. Once that table exists, “when did this break and why is it slow” stops being an archaeology project and becomes a SELECT. This is the complete build: the Snowflake schema, the Airflow callback code that captures context at both ends of every task, and what the whole thing looks like when it runs.

TL;DR

→ DAG-level success/failure is too coarse. Capture context at task start and task end for granular observability — timing, retries, and the exact micro-batch window per task.

→ Airflow exposes the execution context through callbacks: on_execute_callback fires right before a task runs (your “start” hook), and on_success_callback / on_failure_callback fire at the end. Each receives the full context dictionary.

→ The context carries what you need: logical_date (the micro-batch window), dag_run.run_id, ti.try_number, ti.start_date, plus ds/ds_nodash for partition keys. In Airflow 3, access it programmatically with get_current_context() from the Task SDK.

→ Attach the callbacks once via default_args and every task in the DAG is audited automatically — no per-task boilerplate.

→ Ship rows to a centralized Snowflake PIPELINE_AUDIT_LOG table keyed by dag_id + task_id + run_id + try_number, with a START row and an END row per attempt so duration and status fall out of a simple query.

→ Once the data lands, debugging execution delays is a SELECT … ORDER BY duration_seconds DESC, and finding the slowest task in the slowest run is a window function, not a log grep.

Why DAG-level notifications aren’t observability

A DAG success signal answers one coarse question. The unit of observability you actually want is the task attempt.

A DAG success notification answers one question: did the whole thing finish without an unhandled exception? That’s necessary and nowhere near sufficient. It can’t tell you which task in the chain was slow, whether a task silently ran on its second retry, which logical date window each task actually processed, or how today’s run compares to last week’s for the same task. Those are the questions you actually have during an incident, and log-grepping to answer them is how a five-minute diagnosis becomes a two-hour one.

The unit of observability you want is the task attempt, not the DAG run. Every task attempt has a start, an end, a try number, and a logical date. If you record those four things for every attempt in one queryable place, you can answer “when did this get slow,” “which task is the bottleneck,” and “did this run process the window it should have” directly — and you can do it after the fact, without the worker still being alive.

Step 1: the Snowflake audit table

Start with the destination. The schema is deliberately simple — one row per task-attempt per phase (START and END), keyed so you can pair them up and compute duration. Keeping START and END as separate rows (rather than updating one row) means a task that dies hard still leaves its START row behind, which is itself a signal.

CREATE TABLE IF NOT EXISTS ops.pipeline_audit_log (
    audit_id        STRING DEFAULT UUID_STRING(),
    dag_id          STRING       NOT NULL,
    task_id         STRING       NOT NULL,
    run_id          STRING       NOT NULL,
    try_number      NUMBER       NOT NULL,
    phase           STRING       NOT NULL,   -- 'START' | 'END'
    status          STRING,                  -- 'RUNNING' | 'SUCCESS' | 'FAILED'
    logical_date    TIMESTAMP_NTZ,           -- the micro-batch window
    event_time      TIMESTAMP_LTZ DEFAULT CURRENT_TIMESTAMP(),
    duration_sec    NUMBER,                  -- populated on END
    operator        STRING,
    map_index       NUMBER,                  -- for dynamically mapped tasks
    hostname        STRING,
    error_message   STRING,
    loaded_at       TIMESTAMP_LTZ DEFAULT CURRENT_TIMESTAMP()
);

A few deliberate choices. logical_date is stored as its own column because it’s the micro-batch window the task is for — distinct from event_time, the wall-clock moment the row was written. Conflating those two is the single most common audit-table mistake, and it’s exactly the confusion that hid the “wrong window” bug in the opening story. try_number is in the key because retries are first-class events you want to see, not noise to collapse. And map_index is there so dynamically mapped tasks (the .expand() fan-out) each get their own audit trail instead of blurring together.

Step 2: extracting the execution context

Airflow hands you everything through the context dictionary. The pieces that matter for auditing:

def extract_audit_fields(context: dict) -> dict:
    """Pull the audit-relevant fields out of the Airflow context."""
    ti = context["ti"]                     # the TaskInstance
    dag_run = context["dag_run"]

    return {
        "dag_id":       ti.dag_id,
        "task_id":      ti.task_id,
        "run_id":       dag_run.run_id,
        "try_number":   ti.try_number,
        # logical_date is the micro-batch window this run is FOR.
        # Asset-triggered DAGs in Airflow 3 have none — fall back to None.
        "logical_date": context.get("logical_date"),
        "operator":     ti.operator,
        "map_index":    ti.map_index,
        "hostname":     ti.hostname,
        "start_date":   ti.start_date,
    }

The distinction that trips people up: logical_date (formerly execution_date) is the window the run represents, which may be hours or months before the wall clock if you’re backfilling. ti.start_date is when the task actually began executing. You want both — one to know what the task processed, the other to know when and how long. In Airflow 3, if you’re inside task code rather than a callback, you get the same dictionary with from airflow.sdk import get_current_context and context = get_current_context().

Step 3: the callbacks that fire at start and end

This is the heart of it. on_execute_callback runs immediately before the task’s own code — that’s your START row. on_success_callback and on_failure_callback run after — those are your END rows, one carrying SUCCESS, the other FAILED plus the exception.

from datetime import datetime, timezone

def _write_audit_row(fields: dict) -> None:
    """Insert a single audit row into Snowflake via a reusable hook."""
    from airflow.providers.snowflake.hooks.snowflake import SnowflakeHook
    hook = SnowflakeHook(snowflake_conn_id="snowflake_ops")
    hook.run(
        """
        INSERT INTO ops.pipeline_audit_log
            (dag_id, task_id, run_id, try_number, phase, status,
             logical_date, duration_sec, operator, map_index,
             hostname, error_message)
        VALUES
            (%(dag_id)s, %(task_id)s, %(run_id)s, %(try_number)s,
             %(phase)s, %(status)s, %(logical_date)s, %(duration_sec)s,
             %(operator)s, %(map_index)s, %(hostname)s, %(error_message)s)
        """,
        parameters=fields,
    )

def audit_on_start(context: dict) -> None:
    f = extract_audit_fields(context)
    f.update(phase="START", status="RUNNING",
             duration_sec=None, error_message=None)
    _write_audit_row(f)

def audit_on_success(context: dict) -> None:
    f = extract_audit_fields(context)
    duration = (datetime.now(timezone.utc) - f["start_date"]).total_seconds()
    f.update(phase="END", status="SUCCESS",
             duration_sec=round(duration, 2), error_message=None)
    _write_audit_row(f)

def audit_on_failure(context: dict) -> None:
    f = extract_audit_fields(context)
    duration = (datetime.now(timezone.utc) - f["start_date"]).total_seconds()
    f.update(phase="END", status="FAILED",
             duration_sec=round(duration, 2),
             error_message=str(context.get("exception"))[:2000])
    _write_audit_row(f)

Two production notes. First, keep the callback body cheap and defensive — a callback that raises can interfere with task handling, so in a hardened version you wrap _write_audit_row in a try/except that logs and swallows, because a failed audit write should never fail the pipeline. Second, opening a fresh Snowflake connection per callback is fine at low task volume; at high volume you’d batch these through a staging mechanism rather than one INSERT per event, which the “gotchas” section revisits.

Step 4: wire it into every task with one line

The elegance is that you attach these once through default_args, and every task in the DAG inherits them — no per-task decoration, no touching your existing operators.

from airflow import DAG
from airflow.operators.python import PythonOperator
import pendulum

default_args = {
    "on_execute_callback": audit_on_start,
    "on_success_callback": audit_on_success,
    "on_failure_callback": audit_on_failure,
    "retries": 2,
}

with DAG(
    dag_id="sales_etl",
    schedule="@hourly",
    start_date=pendulum.datetime(2026, 1, 1, tz="UTC"),
    catchup=False,
    default_args=default_args,   # <- every task is now audited
) as dag:

    extract = PythonOperator(task_id="extract_orders",
                             python_callable=run_extract)
    transform = PythonOperator(task_id="transform_orders",
                               python_callable=run_transform)
    load = PythonOperator(task_id="load_to_warehouse",
                          python_callable=run_load)

    extract >> transform >> load

That’s the whole integration. Three callbacks defined once, referenced in default_args, and every task — extract, transform, load, and any you add later — writes a START and an END row automatically.

What it looks like when it runs

When the DAG executes, each task emits two rows. Here’s the Airflow task log showing the callbacks firing, followed by the rows that land in Snowflake:

[2026-07-18T02:00:03Z] INFO - Executing on_execute_callback: audit_on_start
[2026-07-18T02:00:03Z] INFO - Audit START written: sales_etl.extract_orders try=1
[2026-07-18T02:00:41Z] INFO - Marking task as SUCCESS. dag_id=sales_etl, task_id=extract_orders
[2026-07-18T02:00:41Z] INFO - Executing on_success_callback: audit_on_success
[2026-07-18T02:00:41Z] INFO - Audit END written: sales_etl.extract_orders try=1 duration=38.4s

And the resulting rows in ops.pipeline_audit_log:

The rows that land in Snowflake. The 112-second transform and the correct 02:00 window are visible at a glance — neither was in the green checkmark.

DAG_ID     TASK_ID          RUN_ID              TRY  PHASE  STATUS   LOGICAL_DATE         DURATION_SEC
---------  ---------------  ------------------  ---  -----  -------  -------------------  ------------
sales_etl  extract_orders   manual__2026-07-18   1   START  RUNNING  2026-07-18 02:00:00        (null)
sales_etl  extract_orders   manual__2026-07-18   1   END    SUCCESS  2026-07-18 02:00:00        38.40
sales_etl  transform_orders manual__2026-07-18   1   START  RUNNING  2026-07-18 02:00:00        (null)
sales_etl  transform_orders manual__2026-07-18   1   END    SUCCESS  2026-07-18 02:00:00       112.65
sales_etl  load_to_warehouse manual__2026-07-18  1   START  RUNNING  2026-07-18 02:00:00        (null)
sales_etl  load_to_warehouse manual__2026-07-18  1   END    SUCCESS  2026-07-18 02:00:00        54.10

Immediately you can see what a green checkmark never showed you: transform_orders took 112 seconds — nearly three times extract — and every task processed the 02:00 logical window as intended. That’s the observability the DAG notification couldn’t give you, and it’s now sitting in a table.

Step 5: the queries that pay it back

The point of the table is what you can ask it. Duration per task-attempt, pairing START and END:

SELECT dag_id, task_id, run_id, try_number,
       MAX(duration_sec) AS duration_sec,
       MAX(CASE WHEN phase = 'END' THEN status END) AS final_status
FROM ops.pipeline_audit_log
GROUP BY dag_id, task_id, run_id, try_number
ORDER BY duration_sec DESC NULLS LAST;
The slowest task in each run — the bottleneck finder — with a window function:

SELECT dag_id, run_id, task_id, duration_sec
FROM (
    SELECT dag_id, run_id, task_id, duration_sec,
           ROW_NUMBER() OVER (PARTITION BY dag_id, run_id
                              ORDER BY duration_sec DESC) AS rn
    FROM ops.pipeline_audit_log
    WHERE phase = 'END'
)
WHERE rn = 1
ORDER BY duration_sec DESC;

And the one that catches silent regressions — a task getting slower over time, comparing each run to that task’s trailing average:

SELECT dag_id, task_id, run_id, logical_date, duration_sec,
       AVG(duration_sec) OVER (
           PARTITION BY dag_id, task_id
           ORDER BY logical_date
           ROWS BETWEEN 7 PRECEDING AND 1 PRECEDING
       ) AS trailing_avg
FROM ops.pipeline_audit_log
WHERE phase = 'END' AND status = 'SUCCESS'
QUALIFY duration_sec > trailing_avg * 1.5   -- 50% slower than usual
ORDER BY logical_date DESC;

That last query is the one that turns the audit log from a forensic tool into an early-warning system: it surfaces the task that’s creeping slower before it becomes the 2 a.m. page.

The gotchas nobody warns you about

A raising callback can disrupt task handling. If audit_on_failure itself throws (say Snowflake is briefly unreachable), you can turn one problem into two. Wrap the write in try/except, log the failure, and swallow it — the audit system must never be able to fail the pipeline it’s observing.

One INSERT per callback will not scale. At a few hundred task-attempts a day it’s fine. At tens of thousands, opening a Snowflake connection per event is both slow and expensive (every connection burns warehouse time). The scalable pattern is to write audit events to a lightweight buffer — a local file, a queue, or Snowpipe/streaming ingestion — and land them in batches, so your observability layer isn’t itself a warehouse cost problem.

try_number semantics shifted across Airflow versions. Historically ti.try_number read differently inside a running task versus after completion, which has burned people building retry logic on it. Pin your understanding to your Airflow version and verify what value you actually get in each callback rather than assuming — a quick log line during rollout saves confusion later.

Asset-triggered DAGs have no logical_date. In Airflow 3, DAGs triggered by asset events don’t get a logical date or the derived ds/ds_nodash variables. Your extract_audit_fields must tolerate None there and lean on dag_run.run_id for identity, or the callback will KeyError on exactly the DAGs you were proud of modernizing.

Wall-clock duration isn’t queue time. The duration computed from ti.start_date is execution time, not the time the task spent waiting in the scheduler queue. If you’re debugging delays specifically, capture the gap between the DAG run’s start and the task’s start too — a task that’s “fast” but starts late points at scheduler or pool contention, a completely different fix than optimizing the task itself.

The one principle

Observability is a table, not a notification. Record every task attempt’s start and end with the execution context Airflow already hands you — logical date, try number, timings — and ship it to one Snowflake table. Then “when did this break, which task is slow, and did it process the right window” become queries instead of log archaeology. A green checkmark tells you nothing failed loudly. An audit row tells you what actually happened — and that’s the difference between hoping your pipeline is healthy and knowing it.

Related reading: Airflow templates & context reference (official docs) · Accessing the Airflow context (Astronomer) · Orchestrating dbt With Airflow on Snowflake · Dynamic Airflow DAGs via Snowflake Metadata · Debugging Zero-Copy Clone Storage Costs in CI/CD