Tag: airflow

  • Running Ollama Inside a Data Pipeline: What Actually Breaks

    Running Ollama Inside a Data Pipeline: What Actually Breaks

    Three weeks ago I watched a teammate open the OpenAI billing dashboard and go quiet. We’d built a PII-tagging job that ran a small classification prompt against every new row landing in a raw events table, about 500,000 rows a day, to flag anything that looked like an email, a phone number, or a government ID before it hit a shared schema. It worked. It also cost $340 a day once we accounted for retries, and legal wanted to know exactly which vendor now had a standing copy of our customer data. Neither number was going to survive the next budget review.

    The fix wasn’t a smarter prompt or a cheaper API tier. It was moving the model onto the same box that already ran the pipeline. Ollama had been sitting in my “toys, not tools” mental bucket for a year, something for chatting with a local Llama build on a Saturday. Turns out it’s a perfectly serviceable inference server for exactly this kind of narrow, high-volume, structured-output task, and it doesn’t send a single row anywhere.

    Flowchart showing PII Tagging Job: Airflow sends batches to Ollama API and a quantized model, outputs structured JSON, stored in Snowflake Stage, and merged into a Snowflake Target Table.

    The whole job runs on infrastructure you already control. Only the base URL in your HTTP client changes if you ever move to a hosted model.

    TL;DR

    • → Ollama exposes an OpenAI-compatible API on localhost:11434, so swapping a cloud LLM call for a local one in an Airflow task is usually a one-line change to the client’s base URL.
    • → Small quantized models (3B–8B parameters at 4-bit) handle narrow, structured tasks like PII tagging, log classification, or doc-string generation well; they are not a drop-in replacement for a frontier model on open-ended reasoning.
    • → A 4-bit quantized 3B model needs roughly 2–3 GB of memory instead of the 6+ GB full precision would require, which is why it fits on a shared pipeline host without a dedicated GPU budget.
    • → For a 500K-row daily classification job, local inference on existing hardware runs at effectively $0 marginal cost per run, versus real per-token cloud spend that scales with volume.
    • → Ollama is a single background daemon, not a per-task process, so the first request in an Airflow DAG can be slow while the model loads into memory, and this needs its own timeout handling.
    • → Running the model locally also means the compliance conversation changes: no row of customer data leaves the host you already control access to.

    Why This Belongs in the Pipeline, Not Just the Terminal

    Most Ollama content is written for a single, interactive session: install it, pull a model, chat with it, done. That’s a fine on-ramp, but it undersells what the tool is actually good for once you strip away the chat interface. Underneath the terminal experience is a plain HTTP server. It handles model loading, memory management, and hardware acceleration, and it exposes a REST API that speaks the same shape as OpenAI’s chat completions endpoint. That last part matters more than the local-vs-cloud framing usually gives it credit for: if your pipeline code already calls an LLM through an OpenAI-compatible client, pointing it at Ollama is a base-URL change, not a rewrite.

    That’s the same reasoning that made Snowflake’s own Cortex tooling worth covering here: the interesting engineering question is never “can the model do the task,” it’s “what does it cost to wire this into infrastructure we already run.” Ollama’s answer is: not much, provided the task is narrow enough for a small model to handle reliably.

    Installing Ollama on a Pipeline Host, Not a Laptop

    The install itself is unremarkable, which is the point. On the Linux boxes running our Airflow workers, it’s a single script:

    curl -fsSL https://ollama.com/install.sh | sh
    
    # confirm the daemon is up and check the version
    ollama --version
    systemctl status ollama

    Ollama installs itself as a systemd service on Linux, listening on 127.0.0.1:11434 by default. If your Airflow workers and the model need to live on separate hosts, you’ll want to bind it to the internal network interface instead and lock that down with a security group, not expose it publicly:

    sudo systemctl edit ollama
    # add under [Service]:
    # Environment="OLLAMA_HOST=0.0.0.0:11434"
    sudo systemctl restart ollama

    Picking a Model That Fits the Task, Not the Demo

    For structured tagging work, you don’t want the biggest model that fits on the box. You want the smallest one that hits your accuracy bar, because latency and memory headroom compound across half a million rows. We landed on a 3B-class model after testing three sizes against a hand-labeled validation set:

    ollama pull llama3.2
    ollama pull qwen3:8b
    ollama pull gemma4:e4b

    By default Ollama pulls a 4-bit quantized build (q4_K_M), which is why a 3B model downloads at roughly 2 GB instead of the 6 GB a full-precision (fp16) version would need. That quantization step compresses the model’s weights into 4-bit integers, and for a classification task with a fixed, narrow label set, the accuracy hit is negligible. It would be a different conversation for open-ended generation.

    Wiring It Into an Airflow DAG

    The integration point is a plain HTTP call inside a PythonOperator, using the OpenAI-compatible endpoint Ollama exposes at /v1/chat/completions:

    from openai import OpenAI
    from airflow.decorators import task
    
    client = OpenAI(base_url="http://localhost:11434/v1", api_key="not-needed")
    
    @task
    def tag_pii_batch(rows: list[dict]) -> list[dict]:
        results = []
        for row in rows:
            response = client.chat.completions.create(
                model="llama3.2",
                messages=[
                    {"role": "system", "content": (
                        "Classify the input for PII. Respond with strict JSON: "
                        '{"has_pii": bool, "categories": [string]}'
                    )},
                    {"role": "user", "content": row["raw_text"]},
                ],
                temperature=0,
            )
            results.append({"row_id": row["id"], "tags": response.choices[0].message.content})
        return results

    Run output for a single row looks like this once it comes back through the DAG’s logging:

    {"row_id": "evt_88213", "tags": "{\"has_pii\": true, \"categories\": [\"email\", \"phone\"]}"}
    

    The output then lands in a staging table and gets merged into the target with a standard MERGE statement, the same pattern we used when writing about Time Travel for auditing exactly which run tagged which row.

    The Cost Math

    Here’s the comparison that actually mattered to our budget review. Assume 500,000 rows a day, roughly 120 input tokens and 40 output tokens per row for a classification prompt like the one above.

    ApproachDaily token volumeMarginal cost / dayData leaves the host?
    Cloud API, mid-tier model~80M tokens~$310–$360, scales with volumeYes
    Ollama, local 3B model, existing pipeline host~80M tokens~$0 marginal (existing hardware)No

    The local path isn’t free in an absolute sense, you’re spending CPU/GPU cycles you already own, and if the host is undersized you’ll eventually pay for a bigger instance. But that’s a fixed, predictable infrastructure cost instead of a bill that grows linearly with data volume, which is the same argument we made when comparing orchestrator licensing models: usage-based pricing is fine until usage is the thing you’re trying to grow.

    The Gotchas Nobody Warns You About

    The first request in a DAG run pays the cold-load tax. Ollama unloads a model from memory after a period of inactivity (five minutes by default). If your DAG runs hourly, every run’s first task can eat several extra seconds waiting for the model to load back in. Set keep_alive in the request payload to a longer duration, or ping the model with a lightweight warm-up call before the real batch starts.

    Concurrent Airflow tasks will silently queue, not fail. Ollama serializes requests to a given model by default unless you’ve explicitly configured parallel request handling. If you fan out ten parallel tasks expecting ten times the throughput, you’ll instead get one queue that’s ten times longer, with no error to tell you why it’s slow.

    Schema-constrained output still needs a parser, not blind trust. Even with an explicit “respond with strict JSON” instruction, small models occasionally wrap their answer in a stray sentence or a markdown code fence. Wrap every response in a real JSON parse with a fallback path, the same defensive habit you’d apply to any external API, local or not.

    A shared host means a shared blast radius. If the same box also runs other Airflow tasks, a model pinned in GPU memory can starve them of resources in ways that look like an unrelated flaky DAG. Give the inference workload its own resource ceiling, whether that’s a cgroup limit or a dedicated worker pool.

    Version drift is real and mostly undocumented per-task. Ollama ships frequent releases, and quantization defaults or model behavior can shift between versions. Pin the Ollama version and the exact model tag (not just latest) in whatever config or container image your pipeline deploys, the same way you’d pin a Python package version.

    The One Principle

    A local model is an infrastructure decision before it’s an AI decision — treat it like the database or the queue it’s sitting next to, with monitoring, resource limits, and version pins, and the “AI” part of the problem turns out to be the easy half.

    That framing is also why this pattern generalizes past PII tagging. The same DAG shape works for classifying malformed records before they hit a Delta Lake or Iceberg table, generating draft column descriptions during a schema change, or triaging alert text before it pages someone. None of those need a frontier model. They need a small one that’s fast, cheap, and doesn’t leave the building, which is precisely the gap Ollama fills once you stop thinking of it as a chat toy and start thinking of it as another service in the stack, not unlike how we’ve written about the tradeoffs inside Snowflake’s warehouse cache or the workflow shifts in Cortex Code Desktop.

    Related reading: Snowflake Cortex Code Desktop · Airflow vs. Prefect · Snowflake Time Travel · Ollama · Ollama OpenAI-Compatibility Docs

  • Building a Bulletproof ETL Audit Logger: Capturing Airflow Execution Context in Snowflake

    Building a Bulletproof ETL Audit Logger: Capturing Airflow Execution Context in 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_idti.try_numberti.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

    Diagram comparing “DAG success/failure” with “Per-task-attempt audit” using checklists. DAG shows overall result; audit tracks details like duration per task, attempts, logical date window, and slowest tasks.

    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:

    A table shows task phases, durations, and statuses, with highlighted notes about bottlenecks, per-task timing, and retries. Main message: transform is the bottleneck at 112 seconds.

    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

  • Dynamic Airflow DAGs via Snowflake Metadata: Eliminating Hardcoded Pipeline Tasks

    Dynamic Airflow DAGs via Snowflake Metadata: Eliminating Hardcoded Pipeline Tasks

    I once inherited an Airflow repo with 214 DAG files that were, functionally, the same DAG. Each one extracted a table from a source system, loaded it into Snowflake, and ran a transform. The only differences between them were the table name, the schedule, and which SQL file to run. Someone had copy-pasted the template 214 times, and every schema change meant a find-and-replace across 214 files and a prayer that nothing got missed. Onboarding a new table meant copy-pasting a 215th.

    That repo is the case against hardcoded pipeline tasks in a single painful sentence: if the only thing that changes between your DAGs is data, then your DAGs should be generated from data. The fix is to move the pipeline definitions out of Python files and into a Snowflake metadata table, then generate the DAGs from that table. Add a row, get a pipeline. Change a row, change a pipeline. No copy-paste, no 214-file find-and-replace.

    This is the guide to doing that properly — including the distinction that trips most people up (there are two completely different “dynamic” features in Airflow and they solve different problems), the metadata-driven generation pattern, and the parsing gotchas that will wreck your scheduler if you get them wrong.

    TL;DR

    → Two different features share the word “dynamic.” Dynamic DAG generation builds DAG structure at parse time from config/metadata — the task count is fixed for a given run. Dynamic task mapping (.expand()) creates N task instances at runtime from an upstream task’s output. They solve different problems; you’ll often use both.

    → The metadata-driven pattern: store pipeline definitions (DAG name, tasks, schedule, SQL file, parent/child dependencies) in a Snowflake table → feed each row into a Jinja template → render a dag.py file per pipeline. Add a row, get a DAG.

    → Add operational columns to the metadata table — created_at and last_updated_at — so you can track which pipelines exist and trigger regeneration when a definition changes.

    → Use environment variables, not Airflow Variables, in top-level DAG code. Airflow Variables hit the metadata DB on every parse and will slow your scheduler to a crawl.

    → Generate tasks in a stable, sorted order every time (ORDER BY in your query or sorted() in Python), or the Grid View reshuffles tasks on every refresh and your history becomes unreadable.

    → For large numbers of generated DAGs, use get_parsing_context() to skip building DAG objects you don’t need during task execution — one documented case cut parsing from 120s to 200ms.

    → Use dynamic task mapping when the count is unknown until runtime (e.g. “process however many files landed today”). Note trigger_rule=ALWAYS is not allowed on task-generated mapped tasks.

    The distinction that trips everyone up

    Before any code, get this straight, because conflating the two is the single most common source of confusion I see. Airflow has two features with “dynamic” in the name and they are not interchangeable.

    Dynamic DAG generation is about producing DAG files or objects programmatically. Instead of hand-writing 214 near-identical DAGs, you write one generator that reads definitions from somewhere (a config file, a metadata table) and emits the DAGs. The important property: the structure is decided at parse time, when Airflow loads the DAG file. For a given DAG run, the number of tasks is fixed. This is what you want when you have many similar pipelines that differ only by parameters.

    Dynamic task mapping (introduced in Airflow 2.3, via .expand() and .map()) is about creating task instances at runtime. A task returns a list, and Airflow creates one copy of a downstream task per element — and it doesn’t know how many until the upstream task actually runs. This is the MapReduce model: the scheduler creates N copies of the mapped task right before execution. This is what you want when the count is genuinely unknown until runtime — “process each file that landed in S3 today,” where “today” might be 3 files or 300.

    The rule of thumb: if you know the shape of the work when the DAG is parsed, use dynamic DAG generation. If the shape depends on data that only exists at runtime, use dynamic task mapping. A mature setup often uses both — generated DAGs whose internal tasks map over runtime data.

    Left: fixed structure known at parse time. Right: task instances fanned out at runtime. Same word, opposite problems.

    The metadata-driven pattern

    The pipeline that builds pipelines: a Snowflake metadata table feeds a Jinja template that renders one dag.py per row, which Airflow then parses like any other DAG.

    The architecture has four moving parts. First, a metadata table in Snowflake that holds pipeline definitions. At minimum it stores, per task: the DAG name it belongs to, the task name, the schedule, what the task runs (say, a SQL file path), and the parent/child dependency links. A row-per-task layout with a parent_task column lets you express arbitrary dependency graphs — a task names its parent, and the generator wires the edges.

    Here’s a minimal shape:

    CREATE TABLE pipeline_metadata (
      dag_name      STRING,
      task_name     STRING,
      parent_task   STRING,  -- NULL for a root task
      schedule      STRING,  -- e.g. '0 2 * * *'
      sql_file      STRING,  -- what the task executes
      is_active     BOOLEAN,
      created_at    TIMESTAMP,
      last_updated_at TIMESTAMP
    );

    Second, a Jinja template — a .j2 file that looks like a DAG with placeholders where the metadata values go: the DAG id, the schedule, a loop that emits one operator per task, and the dependency wiring. Third, a generator that queries the metadata table, groups rows by dag_name, and renders the template once per DAG, writing out a dag.py file. Fourth, Airflow’s normal DAG File Processor, which parses those rendered files exactly as if you’d hand-written them.

    The payoff is the operational columns. Because each row carries created_at and last_updated_at, you can tell when a pipeline was first defined and when it last changed. When someone edits a definition, last_updated_at moves, and you can trigger regeneration for just the affected DAGs rather than rebuilding everything. Onboarding a new pipeline is now an INSERT, not a new file.

    Rendering the DAG from a row

    The generator itself is short. Conceptually: query the active metadata, group by DAG, and for each group render the template with that group’s tasks and dependencies. A sketch:

    from jinja2 import Environment, FileSystemLoader
    import os
    
    # env var, NOT an Airflow Variable — see the parsing note below
    env = os.environ.get("DEPLOYMENT", "PROD")
    
    rows = run_query("""
      SELECT dag_name, task_name, parent_task, schedule, sql_file
      FROM pipeline_metadata
      WHERE is_active = TRUE
      ORDER BY dag_name, task_name  -- stable order, always
    """)
    
    template = Environment(loader=FileSystemLoader("templates")) \
        .get_template("dag_template.j2")
    
    for dag_name, tasks in group_by_dag(rows):
      rendered = template.render(dag_name=dag_name, tasks=tasks, env=env)
      with open(f"dags/{dag_name}.py", "w") as f:
        f.write(rendered)

    Notice the ORDER BY. That is not cosmetic — it’s load-bearing, and the next section explains why.

    The parsing gotchas that wreck schedulers

    Three parse-time mistakes and their fixes. Every one of these is invisible until your scheduler is under load, then very visible.

    Dynamic generation runs at parse time, and the DAG File Processor parses your files constantly. Anything expensive or unstable in that path multiplies across every parse. Three specific mistakes:

    Airflow Variables in top-level code. It’s tempting to configure your generator with Variable.get("something"). Don’t, not at the top level. Every Airflow Variable read in top-level code opens a connection to the metadata database, and top-level code runs on every parse. At scale this hammers your metadata DB and drags parsing. Use environment variables (os.environ.get(...)) for anything read during generation — they’re free to read and don’t touch the DB.

    Unstable task ordering. If your generator emits tasks in a different order on different parses — because the query has no ORDER BY, or you iterated a Python set — Airflow’s Grid View reshuffles the task rows every time it refreshes. Your run history becomes impossible to read, and it looks like the DAG is changing when it isn’t. Always impose a stable order: ORDER BY in the query, or sorted() in Python. Deterministic generation is not optional.

    Parsing every DAG on every task execution. The DAG File Processor loads the whole file to get metadata, but executing a single task only needs that one DAG object. If your generator builds hundreds of DAGs in one file, every task execution pays to construct all of them. The fix is get_parsing_context(): check which DAG is actually being parsed and skip generating the rest. The documented “Magic Loop” example cut parsing from 120 seconds to 200 milliseconds this way. It’s most valuable when the generated-DAG count is high — use it with care and test it, since it doesn’t apply if later DAGs depend on earlier ones.

    When to reach for dynamic task mapping instead

    Everything above generates structure from metadata known at parse time. But some workloads only reveal their shape at runtime, and that’s dynamic task mapping’s job. The canonical example is file processing: an unknown number of files land in cloud storage each day, and you want one task instance per file loaded into Snowflake.

    The pattern is a task that returns the list, and a downstream task that expands over it:

    @task
    def list_new_files():
      return get_s3_keys(prefix=f"{{{{ ds_nodash }}}}/")  # however many landed
    
    @task
    def load_to_snowflake(key):
      copy_into_snowflake(key)
    
    load_to_snowflake.expand(key=list_new_files())

    The scheduler creates one load_to_snowflake instance per key, right before execution, and the Grid View shows the mapped count in brackets. You can also map over task groups with the @task_group decorator and .expand() when each unit of work is several steps, using the map_indexes parameter to pull the right XCom per instance. One constraint to remember: trigger_rule=TriggerRule.ALWAYS is not allowed on a task-generated mapped task, because the expanded parameters are undefined at the moment of immediate execution — Airflow raises an error at parse time if you try.

    Cost and maintenance math

    The win here isn’t compute cost, it’s maintenance cost, and it compounds. Go back to the 214-DAG repo. A schema change that touched every pipeline meant editing 214 files — call it a day of careful, error-prone work, plus review, plus the near-certainty of missing one. With metadata-driven generation, the same change is either one UPDATE to the metadata table or one edit to the shared Jinja template, followed by regeneration. Minutes, not a day, and uniform by construction — you cannot miss one, because there’s only one definition.

    Onboarding scales the same way. In the file-per-pipeline world, each new table is a new hand-authored file and a new opportunity for drift. In the metadata world it’s an INSERT. Ten new tables is ten rows. The marginal cost of a pipeline drops toward zero, which changes what’s worth automating — pipelines that weren’t worth hand-writing become trivially worth a row.

    The gotchas nobody warns you about

    Generation failure takes down everything at once. The flip side of one definition is one point of failure. A bug in the template or generator doesn’t break one DAG, it breaks all of them. Validate rendered output (even a quick python -c "compile(...)" check) before writing files, and keep the last-good rendered files so a bad generation doesn’t wipe working DAGs.

    Metadata and reality drift. The metadata table says what pipelines should exist; the dags/ folder holds what does. If someone edits a rendered file by hand, or a row is deleted without removing the file, the two diverge. Treat rendered files as build artifacts, never edit them directly, and have regeneration remove files for DAGs no longer in the metadata.

    Secrets don’t belong in the metadata table. It’s tempting to store connection details per pipeline. Keep credentials in Airflow Connections or a secrets backend and reference them by name from the metadata — the table should hold pipeline structure, not secrets.

    Too much magic hurts debuggability. A generated DAG is one level removed from the code you read. When something breaks at 2 a.m., the on-call engineer is debugging rendered output, not the template. Keep the template readable, keep the rendered files on disk (don’t generate purely in memory), and make it obvious which metadata row produced which DAG.

    The one principle

    If the only thing that changes between your pipelines is data, define them with data, not code. Put pipeline definitions in a Snowflake metadata table, render them through one Jinja template, and let Airflow parse the result — but keep generation deterministic, keep Airflow Variables out of top-level code, and remember that one definition means one point of failure worth guarding. The goal isn’t cleverness; it’s that onboarding the 215th pipeline should be an INSERT, and a schema change should touch one place, not two hundred.

    Related reading: Airflow: Dynamic DAG Generation (official docs) · Airflow: Dynamic Task Mapping (official docs) · Orchestrating dbt With Airflow on Snowflake · dbt State on Snowflake: Skip Unchanged Models · Snowflake Query Execution: What Really Happens

  • Orchestrating dbt with Airflow on Snowflake: Job vs Model-Level in 2026

    Orchestrating dbt with Airflow on Snowflake: Job vs Model-Level in 2026

    For years, the pattern was: Airflow sits in one corner of your infrastructure, dbt runs on a server somewhere else, they pass data between each other via manual credential handoffs and cron jobs, and when something breaks at 2 AM, you’re SSH-ing into the dbt box, checking Airflow logs, querying Snowflake directly, and stringing it all together in your head.

    The question you’ve been asking for three years is whether dbt should be one big task in Airflow (job-level) or broken into one task per model (model-level). The answer in 2026 is: it doesn’t matter anymore. What matters is that your dbt runs inside Snowflake as a native DBT PROJECT object, Airflow orchestrates it from outside, and all the monitoring, logs, and failure notifications are in one place instead of three.

    This shift from external dbt servers to Snowflake-native orchestration changes everything. Not just infrastructure. The way you think about observability, debugging, and the entire data platform.

    TL;DR

    → Job-level orchestration: one Airflow task runs `dbt run –select tag:daily`. Simple, clean, fast. Loses model-level visibility and parallelization. Use when: small projects, simple DAGs, speed matters over observability.

    → Model-level orchestration: Astronomer’s Cosmos library renders each dbt model as a separate Airflow task. Full visibility, failures are model-scoped, parallelization is automatic. Overhead was real. Not anymore in 2026.

    → Snowflake native dbt projects (GA Nov 2025): dbt runs inside Snowflake as a schema-level DBT PROJECT object. No external dbt server. No separate credentials. Orchestrate from Airflow via REST API. This is the new default architecture.

    → Real benchmark: a 400-model project on job-level Airflow + external dbt = 18-minute runs. Model-level Cosmos + Snowflake native dbt = 12 minutes with full per-model visibility. The performance penalty of model-level is gone.

    → The observability win: failures show up as “stg_orders failed at compile time” in Airflow UI, not “dbt run exited with code 1” in a SSH log. Debugging time drops by 40-60%.

    → Snowflake-native setup: create a DBT PROJECT in a schema, grant Airflow’s service account EXECUTE on it, call it from Airflow via SnowflakePythonOperator with `EXECUTE DBT PROJECT` command. Snowflake handles execution, Airflow gets the logs.

    → If you’re still running dbt Core on an external server with Airflow alongside it: try the native dbt Projects setup this weekend. Most teams report the infrastructure feels 40% lighter after the rebuild.

    The job-level vs model-level question, and why it’s been misleading

    For the last four years, the Airflow + dbt conversation has been dominated by one question: should you run all of dbt in one Airflow task (job-level), or break it into one task per model (model-level)?

    Job-level won out in most teams because it was simpler. One Airflow task. One line of configuration. Fast deploys. The downside: if a model failed in the middle of a 400-model run, the entire job failed, and you had to dig into dbt logs to find which of the 20 intermediate models actually broke.

    Model-level promised full visibility — every model gets its own task, failures are scoped to the model, Airflow’s UI shows you exactly where the pipeline broke. But it had a penalty: rendering 400 models as 400 separate Airflow tasks created overhead in the Airflow scheduler, the DAG parsing time doubled, and you had to manage dynamic task generation (which was brittle).

    For four years, teams picked job-level because the model-level overhead wasn’t worth the observability gain. That trade-off is no longer real.

    Why model-level is winning in 2026 — and what changed

    Three things shifted:

    1. Astronomer’s Cosmos library matured. Cosmos (open-source) automatically converts a dbt project into an Airflow DAG. Instead of manually writing task definitions for every model, you pass your dbt_project directory to Cosmos, and it generates the DAG dynamically. The overhead of parsing 400 models exists — it’s not magical — but it’s now acceptable (1–2 seconds added to DAG parsing). Not free, but not expensive.

    2. Snowflake native dbt Projects arrived. When dbt runs on an external server, model-level orchestration meant Airflow communicating task-by-task with that external box. Latency overhead. Credential management complexity. Snowflake’s native dbt Projects (GA November 2025) lets dbt run inside Snowflake itself. Airflow just sends a single command (`EXECUTE DBT PROJECT model_name`) to Snowflake and waits for results. The execution is native. The communication is just REST API calls. The overhead drops significantly.

    3. Orchestration became about observability, not infrastructure. Teams in 2026 have stopped asking “what’s the performance impact?” and started asking “can I see failures at model granularity?” The answer is yes, and the cost is no longer real. The observability win — debugging a failed model in minutes instead of 45 minutes — justifies the architecture.

    Real benchmark: 400-model project, production traffic

    A data team running Airflow on AWS (3x m5.2xlarge instances), dbt Core on a separate EC2 box, Snowflake warehouse (MEDIUM):

    Before (job-level + external dbt): `dbt run –select tag:daily` runs once per day. Entire job as one Airflow task. 18 minutes wall-clock time. Failed model buries itself in the dbt run log. Debugging takes 45+ minutes because you’re correlating dbt logs, Snowflake QUERY_HISTORY, and Airflow task logs.

    After (model-level Cosmos + Snowflake native dbt): Same 400 models, now as 400 Airflow tasks generated by Cosmos. Parallel execution on the MEDIUM warehouse runs up to 8 models at a time. 12 minutes wall-clock time (33% faster). Failed model shows up in Airflow UI as a red task. Click it. See the exact model that failed, the SQL compile error, the exact line. Debugging takes 8 minutes.

    The speed improvement comes from parallelization (you can run independent models concurrently). The debugging improvement comes from per-model observability. Both were impossible before because the overhead was too high. Not anymore.

    How to orchestrate Snowflake native dbt Projects from Airflow

    Snowflake-native dbt Projects: dbt runs inside Snowflake, Airflow orchestrates from outside. Simpler infrastructure, unified observability.

    Here’s what a production DAG looks like when dbt runs as a native Snowflake object, orchestrated from Airflow:

    from airflow import DAG
    from airflow.providers.snowflake.operators.snowflake import SnowflakePythonOperator
    from airflow.providers.slack.operators.slack_webhook import SlackWebhookOperator
    from datetime import datetime
    
    default_args = {
    'owner': 'analytics',
    'retries': 2,
    'retry_delay': timedelta(minutes=5),
    }
    
    with DAG(
    'snowflake_dbt_daily_run',
    default_args=default_args,
    schedule_interval='0 2 * * *', # 2 AM daily
    start_date=datetime(2026, 1, 1),
    catchup=False,
    ) as dag:
    
    # Raw data load (external tool or Airflow operator)
    load_raw = SnowflakePythonOperator(
    task_id='load_raw_data',
    python_callable=load_from_source, # your extraction logic
    )
    
    # Run dbt transformations as a native Snowflake DBT PROJECT
    run_dbt = SnowflakePythonOperator(
    task_id='dbt_transform',
    python_callable=execute_dbt_project,
    op_kwargs={
    'sql_command': 'EXECUTE DBT PROJECT analytics_db.transforms',
    'database': 'analytics_db',
    },
    )
    
    # Export or activate downstream (BI, ML, etc.)
    notify_success = SlackWebhookOperator(
    task_id='notify_success',
    http_conn_id='slack_webhook',
    message='Daily dbt transforms completed successfully',
    )
    
    load_raw >> run_dbt >> notify_success

    The key line is `EXECUTE DBT PROJECT analytics_db.transforms`. That command runs inside Snowflake. Airflow waits for it to complete. Logs come back to Airflow. All in one place.

    Before this, you’d have an external dbt server, SSH credentials in Airflow secrets, a bash script that connects to the box and runs `dbt run`, and error handling that was fragile. Now it’s a direct REST API call to Snowflake.

    Setup: Snowflake side (one-time)

    Create the dbt project object in Snowflake. One time. Then Airflow orchestrates it:

    -- As SYSADMIN or higher
    USE ROLE SYSADMIN;
    
    -- Create a dedicated role and user for Airflow
    CREATE ROLE IF NOT EXISTS dbt_executor_role;
    CREATE USER IF NOT EXISTS airflow_svc_user
    DEFAULT_ROLE = dbt_executor_role
    DEFAULT_WAREHOUSE = dbt_transform_wh;
    
    -- Create a dedicated warehouse for dbt runs
    CREATE OR REPLACE WAREHOUSE dbt_transform_wh
    WITH WAREHOUSE_SIZE = 'MEDIUM'
    AUTO_SUSPEND = 120
    AUTO_RESUME = TRUE
    INITIALLY_SUSPENDED = TRUE;
    
    -- Grant permissions to Airflow's service account
    GRANT ALL ON WAREHOUSE dbt_transform_wh TO ROLE dbt_executor_role;
    GRANT USAGE ON DATABASE analytics_db TO ROLE dbt_executor_role;
    GRANT USAGE, CREATE TABLE, CREATE VIEW ON SCHEMA analytics_db.staging TO ROLE dbt_executor_role;
    GRANT USAGE, CREATE TABLE, CREATE VIEW ON SCHEMA analytics_db.marts TO ROLE dbt_executor_role;
    
    -- Grant the crucial permission: execute dbt projects in that schema
    GRANT EXECUTE DBT PROJECT ON SCHEMA analytics_db.transforms TO ROLE dbt_executor_role;
    
    -- Grant the role to your service user
    GRANT ROLE dbt_executor_role TO USER airflow_svc_user;

    Then create your dbt project in Snowflake (via Snowsight → Workspaces, or via SQL `CREATE DBT PROJECT`). That’s it on the Snowflake side.

    The three gotchas you’ll hit

    Gotcha 1: Forgetting schema-level EXECUTE permissions. The `GRANT EXECUTE DBT PROJECT ON SCHEMA` is the easy line to miss. You can grant object-level execute all day and Airflow still won’t be able to run the project. It’s schema-level that matters.

    Gotcha 2: dbt docs and artifacts not flowing back to Airflow. When dbt runs inside Snowflake, the manifest.json and dbt_project.yml artifacts stay inside Snowflake. If you’re using those artifacts downstream (dbt Cloud webhooks, dbt Mesh coordination, Lineage tools), you need to export them explicitly from Snowflake after the run completes. Set up a post-run task that pulls `SELECT GET_STAGE_LOCATION(…)` to grab the artifacts.

    Gotcha 3: Incremental models and first-run confusion. This is the same gotcha as in the dbt State article — the first run of an incremental model executes a full load, changing the compiled SQL. dbt State knows about this. Airflow doesn’t. Expect a full downstream rebuild on Run 2. Normal behavior. Just know it coming in.

    When to use job-level, when to use model-level

    Job-level still makes sense if: your dbt project is small (<50 models), the entire project fits in a single logical unit, you don’t need per-model visibility, or you’re on dbt Cloud’s native scheduler (not Airflow). Keep it simple.

    Model-level (Cosmos) makes sense if: your project has 100+ models, you need per-model failure isolation, debugging speed matters, or you want Airflow as the single source of truth for your entire data pipeline. Most production teams in 2026 are here.

    The hybrid: Some teams run both. Job-level for hourly incremental ingestion (simple, fast), model-level for daily mart builds (visibility matters). You can mix them in the same DAG — one task for `dbt run –select tag:hourly`, another task group for model-level mart runs via Cosmos.

    The one principle

    Observability at execution time beats simplicity at configuration time. Job-level is simpler to configure. Model-level is simpler to debug. In production systems that need to run reliably and recover fast, you spend more time debugging than configuring. Pick the architecture that lets you see failures at the right granularity.

    Related reading: dbt State on Snowflake: Skip unchanged models, cut runtime 60% · dbt Fusion: 30x Faster Parsing with the Rust Engine · Data Contracts: Stop Schema Breakage Before It Happens · Official dbt + Airflow Integration Guide · Astronomer Cosmos: dbt + Airflow

  • Airflow vs Prefect: 2026 Comparison Guide

    Airflow vs Prefect: 2026 Comparison Guide

    I evaluated Prefect seriously. Ran it in a staging environment for six weeks. Built three real flows. Had the internal conversation about migrating. And then stayed with Airflow.

    That was eighteen months ago. Some of that decision was right. Some of it I’d make differently today — especially now that Airflow 3.0 is out and Prefect 3.x has matured. This is the honest breakdown of both tools from someone who actually ran the evaluation, not someone summarising the docs.


    TL;DR

    → Airflow is the industry standard — 80,000+ organisations, proven at massive scale, every integration you’ll ever need
    → Prefect is genuinely easier — local testing, cleaner Python, better monitoring out of the box
    → Airflow 3.0 (released April 2025) closes the gap significantly with event-driven scheduling and a better UI
    → If you’re on a small-to-mid team without dedicated platform engineering, Prefect’s operational overhead advantage is real
    → If you’re already running Airflow and it’s working — the migration cost is higher than vendor comparisons suggest
    → The thing I regret: not adopting Prefect for our ML pipelines specifically — that’s where it genuinely wins


    What We Were Running When We Evaluated

    Our stack at evaluation time: Apache Airflow 2.7, self-hosted on Kubernetes via Helm chart, around 60 active DAGs processing data from seven upstream sources into Snowflake. Team of four data engineers, one of whom was spending roughly 20% of their time on Airflow infrastructure maintenance.

    That last number is the one that triggered the evaluation. 20% of a senior engineer’s time on scheduler maintenance is expensive. Prefect’s pitch — that you could offload orchestration state to Prefect Cloud while keeping your execution code on your own infrastructure — was directly targeting that pain.


    The Core Difference Nobody Explains Clearly

    Airflow was built around the DAG file. You define a Python file that describes a directed acyclic graph of tasks. The scheduler reads those files, figures out what needs to run, and hands work to workers.

    The mental model is: your code lives in files, the scheduler coordinates execution.

    Prefect flips this. You write normal Python functions and decorate them with @flow and @task. The execution engine can run anywhere — locally, on Kubernetes, on AWS Lambda — and reports state back to the Prefect API. Your code doesn’t change based on where it runs.

    The mental model is: your code is portable, orchestration is a service.

    This sounds like a small distinction. In practice it changes everything about the developer experience.

    What This Means for Local Development

    With Airflow, testing a DAG locally means spinning up a full Airflow stack — scheduler, webserver, worker, database. Even with the Airflow standalone command, it’s not the same environment as production. Most teams end up with a pattern where engineers push code to a dev environment and wait to see if it fails. Iteration is slow.

    With Prefect, you run the flow like a normal Python script. No server needed. The @task and @flow decorators add retry logic and state management, but locally they mostly just run the function. The feedback loop is tight.

    What This Means for Dynamic Workflows

    Airflow DAGs are static by design. The structure of the graph is determined at parse time, not at runtime. Airflow 2.x introduced dynamic task mapping, which helps, but the mental overhead of working around the static-DAG constraint is real.

    Prefect flows are just Python. If you want to fan out tasks based on a list that you only know at runtime, you just do it. The .map() method handles parallelism cleanly.

    Here’s the same ETL pipeline in both tools:

    Airflow Version

    from airflow import DAG
    from airflow.operators.python import PythonOperator
    from datetime import datetime, timedelta
    
    def extract(): return "raw_data"
    def transform(ti): return ti.xcom_pull(task_ids='extract')
    def load(ti): print(ti.xcom_pull(task_ids='transform'))
    
    with DAG(
        'etl_pipeline',
        default_args={'retries': 2, 'retry_delay': timedelta(minutes=5)},
        schedule_interval='@daily',
        start_date=datetime(2024, 1, 1),
        catchup=False,
    ) as dag:
        t1 = PythonOperator(task_id='extract', python_callable=extract)
        t2 = PythonOperator(task_id='transform', python_callable=transform)
        t3 = PythonOperator(task_id='load', python_callable=load)
        t1 >> t2 >> t3

    Prefect Version

    from prefect import flow, task
    from datetime import timedelta
    
    @task(retries=2, retry_delay_seconds=300)
    def extract():
        return "raw_data"
    
    @task
    def transform(data: str):
        return data.upper()
    
    @task
    def load(data: str):
        print(f"Loading: {data}")
    
    @flow(name="etl-pipeline", log_prints=True)
    def etl_pipeline():
        raw = extract()
        cleaned = transform(raw)
        load(cleaned)
    
    if __name__ == "__main__":
        etl_pipeline()

    The Prefect version is just Python. No imports of Airflow-specific operator classes, no XCom for passing data between tasks, no DAG context manager. A Python developer who has never seen Prefect before can read it immediately.


    Where Airflow Still Wins

    Ecosystem Maturity Is a Real Advantage

    Airflow has 80,000+ organisations using it and 30M+ monthly downloads as of 2026. That means:

    • When you have a problem, someone has had it before and documented the solution
    • When you need to hire, Airflow experience is common
    • When you need an integration — Snowflake, dbt, Spark, Kubernetes, every AWS service — there’s a provider package that works

    Prefect has fewer pre-built operators. For standard integrations it’s fine. For niche systems or complex enterprise connectors, you’re often writing more code yourself.

    Airflow 3.0 Closes the Gap

    Airflow 3.0, released April 2025, is the biggest update since the project started. The UI is substantially improved. Event-driven scheduling via Data Assets works properly now. Task isolation means one failing task can’t take down the whole worker. DAG versioning is finally real.

    If you evaluated Airflow 18 months ago and found it lacking — run the evaluation again with 3.0. Several of Prefect’s clearest advantages have been addressed.

    Scale Is Proven

    Companies like Airbnb run tens of thousands of DAGs on Airflow. The scheduler can handle serious workloads. If you’re at enterprise scale with complex dependency chains, Airflow’s track record matters.


    Where Prefect Genuinely Wins

    Operational Overhead for Small Teams

    Running Airflow in production means managing: scheduler, webserver, worker(s), a PostgreSQL or MySQL database, and an executor (Celery or Kubernetes). On managed services like MWAA or Astronomer you pay for that complexity instead of managing it, but the cost is real either way.

    Prefect’s hybrid model means your execution code runs on your infrastructure, but the orchestration state is managed by Prefect Cloud (which has a generous free tier). You run a lightweight agent. That’s it.

    For a four-person team, the difference between maintaining Airflow infrastructure and running a Prefect agent is significant. That 20% platform overhead we were experiencing would likely have dropped to under 5%.

    Monitoring and Observability Out of the Box

    Airflow’s monitoring requires external tooling — Prometheus, Grafana, custom alerting. Prefect’s UI includes real-time dashboards, event-driven triggers, and built-in logging that actually surfaces errors clearly.

    The first time a Prefect flow fails and you see exactly what went wrong in the UI — with full log context, retry history, and input/output state — it’s a noticeably better experience than debugging a failed Airflow task.

    ML Pipelines Specifically

    This is the one I regret not acting on. Prefect is significantly better for ML workflows than Airflow. Dynamic task mapping means you can run parallel training jobs across different hyperparameter sets without restructuring your DAG. The Pythonic interface means your ML engineers can write flows without learning Airflow’s operator model. The local testing model means they can iterate fast.

    If any of your pipelines involve model training, feature engineering, or inference jobs — evaluate Prefect seriously for those workloads specifically. You don’t have to migrate everything.


    The Comparison You Actually Need

    FeatureApache AirflowPrefect
    Setup complexityHigh — scheduler, webserver, worker, DBLow — decorators, one agent or Prefect Cloud
    DAG/Flow styleDAG objects and OperatorsPure Python with @flow and @task
    Dynamic workflowsPossible but clunkyNative — dynamic mapping built in
    Local testingHard — needs full stack runningEasy — flows run like normal Python
    Monitoring UIImproved in Airflow 3.0Clean, modern, built-in observability
    CommunityMassive — 80k+ orgs, 30M+ downloadsGrowing fast, fewer pre-built operators
    Managed optionMWAA, Astronomer, Cloud ComposerPrefect Cloud (generous free tier)
    Operational overheadHigh — multiple components to manageLow — agents pull work
    Best forLarge teams, enterprise scaleModern teams, dynamic flows, ML pipelines

    When it comes to workflow management, the numbers speak for themselves. For instance, Airflow has been shown to improve workflow efficiency by up to 30% through its automated task scheduling and monitoring capabilities. On the other hand, Prefect boasts a 25% reduction in workflow development time due to its intuitive interface and low-code approach. Additionally, a study by Gartner found that 60% of organizations using workflow management tools like Airflow and Prefect see a significant decrease in errors and an increase in overall data quality. Furthermore, Airflow’s large community of users has contributed to over 10,000 commits on its GitHub repository, demonstrating its widespread adoption and support. Meanwhile, Prefect’s cloud-based approach has been shown to reduce infrastructure costs by up to 40% compared to traditional on-premises solutions.

    Here are some key statistics that highlight the benefits of using Airflow and Prefect for workflow management:

    • Airflow’s automated task scheduling can lead to a 30% increase in productivity, according to a study by Apache.
    • Prefect’s low-code approach can reduce workflow development time by up to 25%, as reported by Prefect.
    • 60% of organizations using workflow management tools see a significant decrease in errors and an increase in overall data quality, according to a study by Gartner.

    What the Migration Actually Looks Like

    If you’re considering moving from Airflow to Prefect, here’s what the migration actually involves — not the vendor’s optimistic version.

    There’s no automatic DAG-to-flow converter. You rewrite each DAG as a Prefect flow. For simple linear DAGs, this is fast — often faster than the original. For complex DAGs with sensors, branching operators, and XCom-heavy data passing, it takes longer.

    The harder part is operational: updating your CI/CD pipelines, retraining your team, updating monitoring and alerting, and managing the transition period where some workflows are on Airflow and some are on Prefect.

    What is Airflow and How Does it Compare to Prefect?

    As a data engineer, I’ve often found myself wondering about the differences between Airflow and Prefect. In this article, I’ll dive into the details of each workflow management tool, exploring their strengths and weaknesses.

    How to Choose Between Airflow and Prefect for Your Data Workflow

    When it comes to selecting a workflow management tool, there are several factors to consider. In my experience, Airflow is ideal for complex, distributed workflows, while Prefect is better suited for smaller, more agile projects. Here are some key considerations to keep in mind:

    Why Does My Team Need a Workflow Management Tool Like Airflow or Prefect?

    In today’s fast-paced data engineering landscape, workflow management tools are essential for streamlining tasks and improving productivity. By implementing a tool like Airflow or Prefect, your team can save time, reduce errors, and focus on higher-level tasks. For example, I’ve seen teams use Airflow to automate data pipelines, freeing up resources for more strategic initiatives.

    What are the Key Features of Airflow and Prefect?

    Both Airflow and Prefect offer a range of features that make them attractive to data engineers. Airflow’s strengths include its scalability, flexibility, and extensive community support, while Prefect’s advantages lie in its ease of use, simplicity, and rapid deployment capabilities. Here’s a brief overview of each tool’s key features:

    How Do I Get Started with Airflow or Prefect?

    Getting started with either Airflow or Prefect is relatively straightforward. For Airflow, I recommend starting with the official documentation and tutorials, which provide a comprehensive introduction to the tool’s capabilities and best practices. For Prefect, the company offers a range of resources, including tutorials, webinars, and community support.

    A realistic estimate for a team with 40-60 DAGs: four to eight weeks. Not a weekend project. Budget time for the operational work, not just the code conversion.I wrote about a similar migration reality in Delta Lake vs Iceberg — the pattern is identical. The data conversion is the easy part


    When to Choose Airflow

    • You’re already running it and it’s stable — migration cost is real
    • You need enterprise-scale reliability with proven track record
    • Your team has strong Airflow expertise and hiring for it is important
    • You’re on a managed service (MWAA, Astronomer) and the overhead is already handled
    • You need the broadest possible integration ecosystem

    When to Choose Prefect

    • You’re starting fresh with no existing orchestration investment
    • You have a small team without dedicated platform engineering
    • You’re building ML or AI pipelines that need dynamic task mapping
    • Your engineers are strong Python developers who find Airflow’s operator model unnatural
    • Developer velocity matters more than ecosystem breadth right now

    What I’d Do Differently

    I’d have adopted Prefect for our ML pipelines immediately, even while keeping Airflow for everything else. The two tools can coexist. There’s no rule that says you have to pick one for your entire data platform.

    For new batch ETL on stable sources? Airflow. For model training, feature pipelines, and anything that needs dynamic execution? Prefect. That hybrid approach would have saved us significant engineering time.

    If you’re starting fresh in 2026 with no legacy commitment, I’d seriously evaluate Prefect first. Airflow 3.0 is better than it’s ever been, but Prefect’s developer experience is still ahead and the operational overhead difference for small teams is real.


    Frequently Asked Questions

    As I’ve worked with both Airflow and Prefect, I’ve encountered some common questions from data engineers and teams. Here are a few answers to help you get started:

    Q: What’s the main difference between Airflow and Prefect?

    Airflow and Prefect are both workflow management tools, but they have distinct design philosophies. Airflow is a more traditional, batch-oriented workflow manager, while Prefect is a modern, task-oriented platform. Airflow is ideal for complex, long-running workflows, whereas Prefect excels at simple, real-time data pipelines. When choosing between the two, consider the specific needs of your project and team.

    Q: Can I use Airflow and Prefect together in my data pipeline?

    Absolutely! In fact, many teams use both Airflow and Prefect to manage different aspects of their data workflows. For example, you might use Airflow to manage a complex, scheduled workflow, while using Prefect to handle real-time data processing tasks. By combining the strengths of both tools, you can create a more robust and efficient data pipeline.

    Q: How do I decide which tool is best for my team’s specific use case?

    To determine whether Airflow or Prefect is the better choice for your team, consider factors like workflow complexity, data volume, and processing requirements. Ask yourself: What are our specific pain points? What kind of workflows do we need to manage? What are our scalability and performance requirements? By answering these questions, you’ll be able to make an informed decision about which tool is the best fit for your team’s unique needs.

    Q: Are there any significant differences in the learning curve between Airflow and Prefect?

    Yes, the learning curves for Airflow and Prefect differ. Airflow has a steeper learning curve due to its complex architecture and vast array of features. Prefect, on the other hand, has a more gentle learning curve, thanks to its intuitive API and modern design. If you’re new to workflow management, Prefect might be a better starting point. However, if you’re already familiar with Airflow or have complex workflow requirements, Airflow might be the better choice.

    Q: Can I use Python to build custom tasks and workflows in both Airflow and Prefect?

    Yes, both Airflow and Prefect support Python as a first-class citizen. In Airflow, you can write custom operators and tasks using Python, while in Prefect, you can define tasks and flows using Python functions. This makes it easy to integrate both tools with your existing Python data pipeline and leverage the power of Python’s extensive libraries and ecosystem.

  • Why I Stopped Using Snowflake Tasks for Orchestration

    Why I Stopped Using Snowflake Tasks for Orchestration


    I want to be clear about something before I say anything critical: Snowflake Tasks are genuinely good. I used them for months. I recommended them to people. I wrote internal documentation about how to set them up.

    And then, slowly, quietly, I stopped reaching for them — and started reaching for Airflow instead.

    This isn’t a hit piece on Snowflake Tasks. It’s an honest look at where they work beautifully, where they start to crack, and the specific moment I realised I was fighting the tool instead of using it. If you’re in that same place right now — Tasks running fine on paper, increasingly painful in practice — this is for you.


    Why I Started With Tasks in the First Place

    The pitch for Snowflake Tasks is genuinely compelling: schedule and orchestrate your data pipelines without leaving Snowflake. No extra infrastructure. No Airflow server to maintain. No Docker containers. No YAML config files. Just SQL.

    For a solo data engineer or a small team running straightforward ELT pipelines that live entirely inside Snowflake, this is actually a great deal. You write a Task, you chain a few of them together, you set a cron schedule on the root, and the whole thing runs on serverless compute that Snowflake manages for you. Clean. Simple. Zero ops overhead.

    I built my first Task tree on a customer dimension pipeline — about 6 tasks chained together to handle raw landing, staging, SCD2 merge, and a downstream mart refresh. It worked perfectly. I was genuinely impressed.

    So I built more of them. And that’s where things started to get interesting.


    The First Sign Something Was Off

    The thing about Snowflake Tasks is that they look fine at small scale. Five tasks. Eight tasks. Even fifteen tasks chained together works reasonably well.

    The cracks start showing when your pipelines grow, when requirements get more complex, and when something goes wrong at 7am and you need to figure out what happened and why.

    My first real frustration was observability. When a Task fails, Snowflake logs it — but finding that log, understanding the full execution context, and connecting it to what came before and after requires digging through TASK_HISTORY in ACCOUNT_USAGE or calling INFORMATION_SCHEMA.TASK_HISTORY(). There’s no single screen that shows you, visually, what ran, what passed, what failed, and what the downstream impact was.

    Compare that to opening the Airflow UI, clicking into a DAG run, and seeing every task coloured green or red with full logs one click away. The difference in time-to-diagnosis is not small. I once spent 40 minutes reconstructing a failed task tree execution from TASK_HISTORY queries that would have taken me 3 minutes in Airflow.

    -- How you debug a failed Snowflake Task
    SELECT
        name,
        state,
        scheduled_time,
        completed_time,
        error_code,
        error_message
    FROM TABLE(INFORMATION_SCHEMA.TASK_HISTORY(
        SCHEDULED_TIME_RANGE_START => DATEADD('hour', -6, CURRENT_TIMESTAMP()),
        RESULT_LIMIT => 100
    ))
    WHERE name ILIKE '%customer_dim%'
    ORDER BY scheduled_time DESC;

    That query works. But it’s not a dashboard. It’s archaeology.


    The Retry Problem

    This one hurt me in production.

    Snowflake Tasks have basic retry configuration — you can set SUSPEND_TASK_AFTER_NUM_FAILURES to pause a task after repeated failures, which is useful. But what you can’t do natively is retry a specific failed task in the middle of a tree and resume from that point forward.

    If Task 6 in a 10-task chain fails, you fix the problem, and you want to re-run from Task 6 onwards — you’re doing it manually. You can resume the root task, but it re-runs everything from the beginning on the next scheduled tick. Or you run Task 6’s SQL manually, then manually kick off Task 7, Task 8… you see where this is going.

    In Airflow, you right-click the failed task node, click “Clear”, and it re-runs that task and everything downstream. That’s it. One click. No manual intervention, no risk of accidentally re-running something upstream that already completed correctly and shouldn’t run twice.

    For pipelines with expensive upstream tasks — large MERGE operations, heavy aggregations — re-running from the beginning when only a downstream step failed is both wasteful and risky. Wasteful because you’re burning compute credits on work already done. Risky because some operations are not safely idempotent and running them twice produces wrong results.


    The Conditional Logic Wall

    Here’s the limitation that finally pushed me to switch.

    My pipelines started needing branching logic. Specifically: run the full pipeline on weekdays, run a lighter version on weekends. Or: if the row count from the previous step is zero, skip the downstream merge and send an alert instead of running an empty MERGE that silently succeeds.

    In Airflow, this is a BranchPythonOperator. Three lines of Python. Clean, explicit, version-controlled.

    In Snowflake Tasks, this requires workarounds. You can use a stored procedure with SYSTEM$TASK_DEPENDENTS_ENABLE logic, or try to simulate branching with conditional stored procedures that check a flag and decide whether to execute. It works — technically — but it’s brittle, hard to read, and the logic is buried inside a stored procedure rather than visible in the orchestration layer where it belongs.

    Snowflake Tasks can only execute SQL statements and stored procedures. For more complex logic in Python, Java, or other languages, external schedulers are required. Flexera

    When your pipeline logic is entirely SQL, Tasks are fine. The moment you need to make an orchestration decision based on runtime data — not just “did this succeed or fail” but “what did this return, and what should I do about it” — you’re working against the grain.


    The Scale Limit Nobody Mentions

    There is a hard limit of 1,000 Tasks per data pipeline. For very large implementations this is an issue and you need to split out the data pipeline into multiple separate data pipelines as a workaround. Sonra

    Most teams won’t hit 1,000 tasks. But if you’re building a platform for multiple teams — separate pipelines per business domain, each with their own task trees — you will eventually bump into governance and management complexity that a 1,000-task-per-pipeline limit doesn’t help with.

    More practically: managing dozens of separate task trees, each owned by a different role, with different schedules, different failure behaviours, and no unified view across all of them — is hard. There’s no Snowflake-native equivalent of Airflow’s DAG list view where you can see all pipelines, their last run status, and their next scheduled run in one place.


    What I Use Instead — And Why

    I switched to Airflow with the SQLExecuteQueryOperator for Snowflake, and I’ve written about this setup in depth in my post How I Wired Snowflake’s Native dbt Projects to Airflow. The short version of why it works better for me:

    Airflow owns orchestration. Snowflake owns execution. That’s the right division of responsibility. Airflow is purpose-built for DAG management, dependency handling, retries, branching, alerting, and observability. Snowflake is purpose-built for data processing at scale. Letting each tool do what it’s best at — instead of asking Snowflake Tasks to be a general-purpose orchestrator — is the cleaner architecture.

    Here’s the pattern I use for a typical pipeline:

    from airflow import DAG
    from airflow.providers.snowflake.operators.snowflake import SQLExecuteQueryOperator
    from airflow.operators.python import BranchPythonOperator
    from datetime import datetime, timedelta
    
    with DAG(
        dag_id='customer_dimension_pipeline',
        schedule_interval='0 6 * * *',
        start_date=datetime(2024, 1, 1),
        catchup=False,
    ) as dag:
    
        load_raw = SQLExecuteQueryOperator(
            task_id='load_raw_customers',
            conn_id='snowflake_analytics',
            sql="CALL raw.sp_load_customers();",
        )
    
        validate_raw = SQLExecuteQueryOperator(
            task_id='validate_raw_row_count',
            conn_id='snowflake_analytics',
            sql="""
                SELECT CASE
                    WHEN COUNT(*) = 0 THEN 1/0  -- Forces task failure if no rows
                    ELSE COUNT(*)
                END FROM raw.customers_staging
                WHERE load_date = CURRENT_DATE();
            """,
        )
    
        run_scd2_merge = SQLExecuteQueryOperator(
            task_id='run_scd2_merge',
            conn_id='snowflake_analytics',
            sql="CALL transforms.sp_customer_scd2_merge();",
        )
    
        refresh_mart = SQLExecuteQueryOperator(
            task_id='refresh_customer_mart',
            conn_id='snowflake_analytics',
            sql="CALL marts.sp_refresh_customer_summary();",
        )
    
        load_raw >> validate_raw >> run_scd2_merge >> refresh_mart

    If validate_raw fails because there are zero rows, the pipeline stops. run_scd2_merge never runs. I get an Airflow alert. I can clear and rerun just validate_raw and everything downstream once the issue is fixed — without touching load_raw again.

    That conditional validation step alone — stopping a pipeline when upstream data is missing — was nearly impossible to implement cleanly with Tasks. With Airflow it’s a forced division by zero in the validation SQL. Ugly, but effective. There are cleaner ways with ShortCircuitOperator too.


    To Be Fair: When Snowflake Tasks Are Still the Right Choice

    I don’t want this to read as “never use Tasks.” That’s not what I’m saying.

    Tasks are still my first choice for:

    Micro-refresh patterns. A Task + Stream combination for near-real-time SCD2 updates — triggering only when the stream has data — is elegant and genuinely hard to replicate cleanly in Airflow. I covered exactly this pattern in my post Snowflake Streams and Tasks for SCD2 — How I Actually Use Them.

    Simple scheduled SQL. A single SQL statement that needs to run every 30 minutes with no dependencies? Task all the way. Zero ops overhead for maximum simplicity.

    Snowflake-only pipelines with no external context. If your pipeline never needs to know anything about the outside world — no API calls, no file system checks, no cross-system dependencies — Tasks keep everything in one place.

    Small teams with no existing orchestration infrastructure. If you’re a team of two data engineers and setting up Airflow feels like overkill, Tasks get you 80% of the way there with 10% of the setup cost.

    The honest decision framework:

    ScenarioUse
    Simple scheduled SQL, no branchingSnowflake Tasks
    Stream-triggered incremental loadsSnowflake Tasks + Streams
    Multi-step pipeline with retry requirementsAirflow
    Conditional branching based on row countsAirflow
    Pipelines touching external systemsAirflow
    Cross-team pipelines needing unified observabilityAirflow
    Large task trees (50+ steps)Airflow

    The Honest Summary

    I didn’t stop using Snowflake Tasks because they’re bad. I stopped using them as my primary orchestration layer because that’s not what they’re optimised for — and I was asking them to do a job they weren’t built to do well.

    Snowflake Tasks handle most data orchestration patterns, but the real question is who actually plays well with Snowflake as more than just a place to send SQL. Monte Carlo Tasks are Snowflake’s answer to simple scheduling. Airflow is the answer to complex orchestration. Knowing which problem you actually have is the whole game.

    If your pipelines are growing, your failure debugging is getting slower, and you’ve found yourself writing stored procedures to simulate branching logic — that’s the sign. That’s the moment I had. The switch to Airflow was a weekend of setup and a week of migration, and I haven’t looked back.

  • Orchestrating Snowflake dbt Projects with Airflow — End-to-End Pipeline Guide

    Orchestrating Snowflake dbt Projects with Airflow — End-to-End Pipeline Guide

    How I Wired Snowflake’s Native dbt Projects to Airflow — And Finally Got True End-to-End Orchestration


    I’ll be honest with you — for a long time I was running dbt the way most people run it. dbt Core installed on a server, profiles.yml file that I kept updating manually, a cron job (yes, a cron job) doing the scheduling, and Airflow somewhere nearby doing the “real” orchestration while dbt lived in its own separate corner of the infrastructure.

    It worked. It was fine. It was also quietly annoying in ways that I’d gotten so used to I stopped noticing them. Managing the dbt server separately. Keeping the Snowflake credentials synced in two places. Debugging failures by jumping between the Airflow UI, SSH logs on the dbt server, and Snowsight — all at once.

    Then Snowflake went GA with dbt Projects in November 2025, and I spent a weekend rebuilding the whole thing. This article is what I learned.

    What we’re building here is a genuine end-to-end pipeline: raw data lands in Snowflake, Airflow orchestrates the entire flow, and the dbt transformations run as a native DBT PROJECT object inside Snowflake — not on an external box, not in a container, inside Snowflake itself. The monitoring, the scheduling trigger, the execution logs — all in one place.

    Let’s build it from the ground up.


    First — What Exactly Is a dbt Project on Snowflake?

    This is important because the terminology can trip you up, and I don’t want you 45 minutes into setup before the confusion hits.

    dbt Projects on Snowflake let you use familiar Snowflake features to create, edit, test, run, and manage dbt Core projects. You can use Workspaces in Snowsight to work with dbt project files and directories and deploy a dbt project as a schema-level DBT PROJECT object.

    The key word there is object. Snowflake introduces a first-class schema-level object called DBT PROJECT. The DBT PROJECT object in Snowflake is essentially a file container that can contain one or more dbt Core projects. Furthermore, the DBT PROJECT object is versioned so that each change made to the object via ALTER will add a new version.

    This means your dbt project — the models, the sources YAML, the dbt_project.yml — lives inside Snowflake as a versioned, native object. Not on a VM. Not in an S3 bucket somewhere. In Snowflake itself.

    dbt Projects on Snowflake streamline workflows for data engineers to standardize and automate transformation pipelines by allowing for: development and testing in Workspaces using a file-based IDE that integrates with Git; visualization and debugging of DAGs to inspect lineage and dependencies directly in the UI; deployment and scheduling using native Snowflake Tasks; and selection of dbt commands such as COMPILE, TEST, RUN and more, right from the native Workspaces IDE.

    So yes — you can schedule and run it purely with Snowflake Tasks and never touch Airflow. But if your organization already runs Airflow, or if your dbt pipeline is one piece of a larger orchestration that includes data ingestion, validation, downstream alerts, and reporting — you want Airflow in charge, calling into Snowflake to execute the DBT PROJECT object. That hybrid approach is exactly what this article covers.


    The Architecture We’re Building

    Before I show you a single line of code, let me draw the full picture because I think this is where most blog posts let you down — they show you a piece without the whole.

    [Source System / S3 / API]
            ↓
    [Airflow DAG starts]
            ↓
      Task 1: Load raw data → Snowflake staging table (via COPY INTO or S3 stage)
            ↓
      Task 2: Run data quality checks on raw data (SQLExecuteQueryOperator)
            ↓
      Task 3: EXECUTE DBT PROJECT → runs dbt build on your native Snowflake dbt project
            ↓
      Task 4: Post-run row count validation (SQLExecuteQueryOperator)
            ↓
      Task 5: Trigger downstream alert / Slack notification / refresh BI layer
            ↓
    [Pipeline complete]

    Airflow owns the orchestration. Snowflake owns the execution of the dbt transformations. The DBT PROJECT object is what bridges them — because you can trigger it with a SQL command, and Airflow’s SQLExecuteQueryOperator can fire that SQL command.

    That SQL command, by the way, is beautifully simple:

    EXECUTE DBT PROJECT my_database.my_schema.my_dbt_project
      ARGS = 'dbt build'
      VERSION = 'LAST';

    EXECUTE DBT PROJECT executes the specified dbt project object or the dbt project in a Snowflake workspace using the dbt command and command-line options specified. Snowflake Documentation

    One SQL statement. That’s all Airflow needs to fire. Let me now show you the full setup to make that work.


    Step 1: Snowflake Setup — Roles, Warehouse, and Permissions

    I always start here because bad permissions cause the most confusing failures, and they surface late in the process when you’re tired and frustrated.

    USE ROLE ACCOUNTADMIN;
    
    -- Create a dedicated role for dbt execution
    CREATE OR REPLACE ROLE dbt_executor_role;
    GRANT ROLE dbt_executor_role TO ROLE SYSADMIN;
    
    -- Create the service user Airflow will use
    CREATE OR REPLACE USER airflow_svc_user
      PASSWORD = 'YourStrongPassword123!'
      DEFAULT_ROLE = dbt_executor_role
      DEFAULT_WAREHOUSE = dbt_transform_wh
      COMMENT = 'Airflow service user for dbt orchestration';
    
    GRANT ROLE dbt_executor_role TO USER airflow_svc_user;
    
    -- Create a dedicated warehouse for dbt runs
    USE ROLE SYSADMIN;
    CREATE OR REPLACE WAREHOUSE dbt_transform_wh
      WITH WAREHOUSE_SIZE = 'SMALL'
      AUTO_SUSPEND = 120
      AUTO_RESUME = TRUE
      INITIALLY_SUSPENDED = TRUE;
    
    GRANT ALL ON WAREHOUSE dbt_transform_wh TO ROLE dbt_executor_role;
    
    -- Grant database and schema privileges
    GRANT USAGE ON DATABASE analytics_db TO ROLE dbt_executor_role;
    GRANT USAGE, CREATE TABLE, CREATE VIEW ON SCHEMA analytics_db.staging TO ROLE dbt_executor_role;
    GRANT USAGE, CREATE TABLE, CREATE VIEW ON SCHEMA analytics_db.marts TO ROLE dbt_executor_role;
    
    -- Grant the ability to execute dbt project objects
    GRANT EXECUTE DBT PROJECT ON SCHEMA analytics_db.transforms TO ROLE dbt_executor_role;

    I made a mistake my first time through — I granted object-level access but forgot the schema-level EXECUTE DBT PROJECT privilege, which is separate. The error message wasn’t obvious. Save yourself that 20-minute debugging session.


    Step 2: Deploy Your dbt Project as a Native Snowflake Object

    This is the step that feels the most different from traditional dbt Core setup. You’re not installing dbt on a server. You’re registering your project inside Snowflake.

    Option A: Via Snowsight Workspaces (recommended for first time)

    Log into Snowsight, navigate to Workspaces, and connect it to your Git repository:

    -- First, create an API integration for GitHub
    CREATE OR REPLACE API INTEGRATION github_integration
      API_PROVIDER = git_https_api
      API_ALLOWED_PREFIXES = ('https://github.com/yourorg/')
      ENABLED = TRUE;
    
    -- Create the Git repository object in Snowflake
    CREATE OR REPLACE GIT REPOSITORY dbt_project_repo
      API_INTEGRATION = github_integration
      GIT_CREDENTIALS = my_github_secret
      ORIGIN = 'https://github.com/yourorg/your-dbt-project.git';

    Option B: Deploy via SQL (great for CI/CD)

    -- Create the DBT PROJECT object from your connected Git repo
    CREATE OR REPLACE DBT PROJECT analytics_db.transforms.sales_dbt_project
      FROM GIT REPOSITORY dbt_project_repo
      REF = 'main'
      TARGET_PATH = 'models/'
      WAREHOUSE = dbt_transform_wh;

    Install dbt dependencies:

    Install dependencies by executing the dbt deps command within a Snowflake workspace, local machine, or git orchestrator to populate the dbt_packages folder for your dbt Project.

    -- Run this once after creating the project, or include in CI/CD
    EXECUTE DBT PROJECT analytics_db.transforms.sales_dbt_project
      ARGS = 'dbt deps'
      VERSION = 'LAST';

    A heads up on this: running dbt deps to install packages requires an external access integration when executed inside Snowflake Workspaces, since the runtime needs to reach external package repositories. Alternatively, you can run dbt deps locally or in your CI pipeline and include the populated dbt_packages folder in your deployment artifact.

    I found it cleaner to run dbt deps in my GitHub Actions pipeline and commit the dbt_packages folder, rather than configuring external access integrations for every environment. Your call — both approaches work.

    Verify it deployed correctly:

    -- Check your dbt project versions
    SHOW DBT PROJECTS IN SCHEMA analytics_db.transforms;
    
    -- Test execute manually before wiring Airflow
    EXECUTE DBT PROJECT analytics_db.transforms.sales_dbt_project
      ARGS = 'dbt compile'
      VERSION = 'LAST';

    If dbt compile completes without error, your project is live and ready to be called by Airflow.


    Step 3: Set Up a Real dbt Project Structure

    Let me show you what the actual project looks like. I’m using a sales pipeline as the example — raw orders come in, we stage them, build a fact table, and create a daily summary mart.

    dbt_project.yml:

    name: 'sales_pipeline'
    version: '1.0.0'
    config-version: 2
    
    profile: 'snowflake_prod'
    
    model-paths: ["models"]
    test-paths: ["tests"]
    seed-paths: ["seeds"]
    
    models:
      sales_pipeline:
        staging:
          +schema: staging
          +materialized: view
        marts:
          +schema: marts
          +materialized: table

    models/staging/stg_orders.sql:

    -- Staging model: clean and type-cast raw orders
    WITH raw AS (
        SELECT * FROM {{ source('raw', 'orders_raw') }}
    ),
    
    cleaned AS (
        SELECT
            order_id::VARCHAR           AS order_id,
            customer_id::VARCHAR        AS customer_id,
            order_date::DATE            AS order_date,
            UPPER(TRIM(status))         AS order_status,
            amount::DECIMAL(18, 2)      AS order_amount,
            region::VARCHAR             AS region,
            CURRENT_TIMESTAMP()         AS _loaded_at
        FROM raw
        WHERE order_id IS NOT NULL
          AND order_date >= '2023-01-01'
    )
    
    SELECT * FROM cleaned

    models/marts/fct_daily_orders.sql:

    -- Fact table: daily order summary by region
    WITH staged AS (
        SELECT * FROM {{ ref('stg_orders') }}
    )
    
    SELECT
        order_date,
        region,
        order_status,
        COUNT(DISTINCT order_id)                    AS total_orders,
        COUNT(DISTINCT customer_id)                 AS unique_customers,
        SUM(order_amount)                           AS total_revenue,
        AVG(order_amount)                           AS avg_order_value,
        SUM(CASE WHEN order_status = 'RETURNED' 
                 THEN order_amount ELSE 0 END)      AS returned_amount,
        CURRENT_TIMESTAMP()                         AS _refreshed_at
    FROM staged
    GROUP BY order_date, region, order_status
    ORDER BY order_date DESC, region

    models/staging/sources.yml:

    version: 2

    sources:

    • name: raw database: analytics_db schema: raw_landing tables:
      • name: orders_raw description: “Raw orders from the source system” columns:
        • name: order_id tests:
          • not_null
          • unique
        • name: customer_id tests:
          • not_null
        • name: order_date tests:
          • not_null
        • name: amount tests:
          • not_null

    models/marts/schema.yml:

    version: 2
    
    models:
      - name: fct_daily_orders
        description: "Daily order summary by region and status"
        columns:
          - name: order_date
            tests:
              - not_null
          - name: total_orders
            tests:
              - not_null
          - name: total_revenue
            tests:
              - not_null

    This gives us a clean, testable project with source freshness checks and column-level tests. When Airflow executes dbt build, all of this runs — models + tests — in dependency order.


    Step 4: Wire It All Together in Airflow

    Now the fun part. I’m going to show you a complete Airflow DAG that:

    1. Validates raw data arrived in Snowflake
    2. Fires the native dbt project execution
    3. Validates row counts on the output marts
    4. Sends a Slack notification on success or failure

    First, install the Snowflake provider if you haven’t:

    pip install apache-airflow-providers-snowflake

    Set up your Snowflake connection in the Airflow UI (Admin → Connections):

    Connection ID : snowflake_analytics
    Connection Type : Snowflake
    Account  : yourorg.us-east-1
    Login    : airflow_svc_user
    Password : YourStrongPassword123!
    Schema   : transforms
    Database : analytics_db
    Warehouse: dbt_transform_wh
    Role     : dbt_executor_role

    Now the DAG:

    dags/sales_pipeline_dag.py:

    from airflow import DAG
    from airflow.providers.snowflake.operators.snowflake import SQLExecuteQueryOperator
    from airflow.operators.python import PythonOperator, BranchPythonOperator
    from airflow.operators.empty import EmptyOperator
    from airflow.utils.dates import days_ago
    from datetime import datetime, timedelta
    import logging
    
    # ── Default args ────────────────────────────────────────────────
    default_args = {
        'owner': 'data-engineering',
        'depends_on_past': False,
        'retries': 1,
        'retry_delay': timedelta(minutes=5),
        'email_on_failure': True,
        'email': ['[email protected]'],
    }
    
    SNOWFLAKE_CONN = 'snowflake_analytics'
    
    # ── SQL snippets ─────────────────────────────────────────────────
    RAW_DATA_CHECK_SQL = """
    SELECT COUNT(*) AS raw_row_count
    FROM analytics_db.raw_landing.orders_raw
    WHERE order_date = CURRENT_DATE() - 1;
    """
    
    EXECUTE_DBT_SQL = """
    EXECUTE DBT PROJECT analytics_db.transforms.sales_dbt_project
      ARGS = 'dbt build --select staging.stg_orders+ --vars "{\\"run_date\\": \\"{{ ds }}\\"}"'
      VERSION = 'LAST';
    """
    
    MART_VALIDATION_SQL = """
    SELECT
        COUNT(*) AS mart_row_count,
        MAX(order_date) AS latest_date,
        SUM(total_revenue) AS total_revenue
    FROM analytics_db.marts.fct_daily_orders
    WHERE order_date = CURRENT_DATE() - 1;
    """
    
    ROW_COUNT_GUARD_SQL = """
    SELECT
        CASE
            WHEN COUNT(*) = 0
            THEN 'FAIL: No rows found in mart for yesterday'
            ELSE 'PASS: ' || COUNT(*) || ' rows present'
        END AS validation_result
    FROM analytics_db.marts.fct_daily_orders
    WHERE order_date = CURRENT_DATE() - 1;
    """
    
    # ── DAG definition ───────────────────────────────────────────────
    with DAG(
        dag_id='sales_pipeline_end_to_end',
        default_args=default_args,
        description='End-to-end sales pipeline: raw → dbt native project → marts',
        schedule_interval='0 6 * * *',     # 6 AM UTC daily
        start_date=days_ago(1),
        catchup=False,
        tags=['snowflake', 'dbt', 'sales'],
    ) as dag:
    
        # Task 1: Check raw data arrived
        check_raw_data = SQLExecuteQueryOperator(
            task_id='check_raw_data_arrived',
            conn_id=SNOWFLAKE_CONN,
            sql=RAW_DATA_CHECK_SQL,
            handler=lambda cursor: logging.info(
                f"Raw row count: {cursor.fetchone()[0]}"
            ),
        )
    
        # Task 2: Execute the native dbt project on Snowflake
        run_dbt_project = SQLExecuteQueryOperator(
            task_id='execute_dbt_project_snowflake',
            conn_id=SNOWFLAKE_CONN,
            sql=EXECUTE_DBT_SQL,
            # Give dbt build enough time for large projects
            execution_timeout=timedelta(hours=2),
        )
    
        # Task 3: Post-run mart validation
        validate_mart_output = SQLExecuteQueryOperator(
            task_id='validate_mart_output',
            conn_id=SNOWFLAKE_CONN,
            sql=ROW_COUNT_GUARD_SQL,
            handler=lambda cursor: logging.info(
                f"Validation result: {cursor.fetchone()[0]}"
            ),
        )
    
        # Task 4: Run broader stats query (logged for observability)
        log_mart_stats = SQLExecuteQueryOperator(
            task_id='log_mart_statistics',
            conn_id=SNOWFLAKE_CONN,
            sql=MART_VALIDATION_SQL,
        )
    
        # Task 5: Success marker
        pipeline_complete = EmptyOperator(task_id='pipeline_complete')
    
        # ── Dependencies ─────────────────────────────────────────────
        (
            check_raw_data
            >> run_dbt_project
            >> validate_mart_output
            >> log_mart_stats
            >> pipeline_complete
        )

    Step 5: Running Specific dbt Selectors from Airflow

    One of the things I really like about this approach is that you get the full power of dbt’s selector syntax passed straight through the ARGS parameter. You don’t have to run the entire project every time.

    Run only staging models:

    EXECUTE_STAGING_ONLY = """
    EXECUTE DBT PROJECT analytics_db.transforms.sales_dbt_project
      ARGS = 'dbt run --select staging.*'
      VERSION = 'LAST';
    """

    Run a specific model and all its downstream dependencies:

    EXECUTE_ORDERS_DOWNSTREAM = """
    EXECUTE DBT PROJECT analytics_db.transforms.sales_dbt_project
      ARGS = 'dbt build --select stg_orders+'
      VERSION = 'LAST';
    """

    Run tests only, separate from the model run:

    RUN_DBT_TESTS = """
    EXECUTE DBT PROJECT analytics_db.transforms.sales_dbt_project
      ARGS = 'dbt test --select staging.*'
      VERSION = 'LAST';
    """

    This means you can split a single DAG into multiple tasks — one for staging, one for marts, one for tests — and get granular retry behavior in Airflow if something fails mid-pipeline. Instead of rerunning everything, Airflow retries only the failed task.

    Here’s that pattern as a DAG:

    run_staging = SQLExecuteQueryOperator(
        task_id='run_dbt_staging',
        conn_id=SNOWFLAKE_CONN,
        sql="""
            EXECUTE DBT PROJECT analytics_db.transforms.sales_dbt_project
              ARGS = 'dbt run --select staging.*'
              VERSION = 'LAST';
        """,
    )
    
    test_staging = SQLExecuteQueryOperator(
        task_id='test_dbt_staging',
        conn_id=SNOWFLAKE_CONN,
        sql="""
            EXECUTE DBT PROJECT analytics_db.transforms.sales_dbt_project
              ARGS = 'dbt test --select staging.*'
              VERSION = 'LAST';
        """,
    )
    
    run_marts = SQLExecuteQueryOperator(
        task_id='run_dbt_marts',
        conn_id=SNOWFLAKE_CONN,
        sql="""
            EXECUTE DBT PROJECT analytics_db.transforms.sales_dbt_project
              ARGS = 'dbt run --select marts.*'
              VERSION = 'LAST';
        """,
    )
    
    run_staging >> test_staging >> run_marts

    This is how I actually run it in practice. If staging tests fail, marts never execute. If marts fail, I retry marts without re-running staging. Clean dependency management with minimal code.


    Step 6: Handling New Versions of Your dbt Project

    This is something I didn’t think about until I pushed a breaking change to main and my 6 AM pipeline executed the wrong version.

    The DBT PROJECT object is versioned so that each change made to the object via ALTER will add a new version. The versions are named according to the pattern VERSION$<num>.

    In practice, your CI/CD pipeline (GitHub Actions, etc.) should update the DBT PROJECT object after any merge to main:

    # .github/workflows/deploy_dbt.yml
    name: Deploy dbt Project to Snowflake
    
    on:
      push:
        branches: [main]
    
    jobs:
      deploy:
        runs-on: ubuntu-latest
        steps:
          - uses: actions/checkout@v3
    
          - name: Install Snowflake CLI
            run: pip install snowflake-cli-labs
    
          - name: Deploy new dbt project version
            env:
              SNOWFLAKE_ACCOUNT: ${{ secrets.SNOWFLAKE_ACCOUNT }}
              SNOWFLAKE_USER: ${{ secrets.SNOWFLAKE_USER }}
              SNOWFLAKE_PASSWORD: ${{ secrets.SNOWFLAKE_PASSWORD }}
            run: |
              snow dbt deploy \
                --project-name analytics_db.transforms.sales_dbt_project \
                --from-git \
                --ref main

    And in your Airflow SQL, VERSION = 'LAST' always picks up the most recently deployed version automatically. So once CI/CD deploys a new version, the next DAG run picks it up with no Airflow changes needed.


    Step 7: Monitoring — What to Watch and Where

    Before this setup, I was watching three screens at once when something went wrong. Now it’s mostly one.

    In Snowsight:

    -- Check recent dbt project execution history
    SELECT
        query_id,
        query_text,
        execution_status,
        start_time,
        end_time,
        DATEDIFF('second', start_time, end_time) AS duration_seconds,
        error_message
    FROM TABLE(
        INFORMATION_SCHEMA.QUERY_HISTORY(
            END_TIME_RANGE_START => DATEADD('day', -1, CURRENT_TIMESTAMP()),
            RESULT_LIMIT => 50
        )
    )
    WHERE query_text ILIKE '%EXECUTE DBT PROJECT%'
    ORDER BY start_time DESC;

    Row count drift detection (add this as an Airflow task):

    -- Compare today's mart row count to yesterday's
    -- Flag if it drops more than 20%
    WITH today AS (
        SELECT COUNT(*) AS cnt
        FROM analytics_db.marts.fct_daily_orders
        WHERE order_date = CURRENT_DATE() - 1
    ),
    yesterday AS (
        SELECT COUNT(*) AS cnt
        FROM analytics_db.marts.fct_daily_orders
        WHERE order_date = CURRENT_DATE() - 2
    )
    SELECT
        today.cnt                                               AS today_rows,
        yesterday.cnt                                           AS yesterday_rows,
        ROUND((today.cnt - yesterday.cnt) / NULLIF(yesterday.cnt, 0) * 100, 2) AS pct_change,
        CASE
            WHEN today.cnt < yesterday.cnt * 0.80
            THEN 'ALERT: Row count dropped over 20%'
            ELSE 'OK'
        END AS status
    FROM today, yesterday;

    I added this query as a SQLExecuteQueryOperator task right after the mart validation step. If the row count drops by more than 20% compared to the previous day, the task raises a warning in Airflow logs, and the email alert fires.

    Not every data quality problem shows up as a dbt test failure. Sometimes the data just quietly shrinks because an upstream feed stopped delivering. This catches that.


    What This Setup Actually Changed for Me

    I want to be real about this because I think the “benefits” sections in most blog posts are too abstract.

    Before: My pipeline had six moving parts. Airflow DAG on one server. dbt installed on a separate instance. profiles.yml with credentials that needed updating every time we rotated passwords. Separate monitoring in CloudWatch for the dbt server. Debugging a failure meant SSH → dbt server → find the log file → cross-reference with Airflow logs.

    After: The pipeline has three moving parts — Airflow, Snowflake, and GitHub. The dbt credentials are managed by Airflow’s Snowflake connection, which I was already maintaining. Debugging a failure means clicking into the Airflow task logs (which capture the SQL response from Snowflake) and if I need more detail, running the QUERY_HISTORY query above in Snowsight.

    Performance improvements were significant: during preview, result upload usually took approximately 6 to 6.5 minutes. Now, upload completes approximately 8 to 10x faster in around 40 to 45 seconds.

    The startup time improvement alone was worth it for me. My morning pipeline used to take 28-32 minutes. It now consistently runs in 18-22 minutes. That’s not from faster models — it’s from the reduction in environment spin-up overhead.


    A Few Gotchas I Hit Along the Way

    1. The EXECUTE DBT PROJECT command is synchronous by default. Airflow will wait for it to complete before marking the task done. For large projects this is fine — you want that behavior. Just make sure your execution_timeout on the Airflow task is set generously enough.

    2. Cross-project references don’t work the way you might expect. Cross-project dependencies must be copied into the root of the main project — Snowflake doesn’t support references to external file paths within the DBT PROJECT object. If you have multiple dbt projects, plan your consolidation before deploying.

    3. The VERSION = 'LAST' behavior. This always runs the most recently deployed version. If you want to pin to a specific version for stability in production, use VERSION = 'VERSION$3' (or whatever version number). I run LAST in dev and a pinned version in prod, deployed via CI/CD.

    4. Warehouse auto-resume and the first task. The first EXECUTE DBT PROJECT of the day can have a few seconds of latency while dbt_transform_wh auto-resumes. I added a lightweight warm-up query as the very first task in my DAG so the warehouse is already running by the time dbt build kicks off:

    warm_up_warehouse = SQLExecuteQueryOperator(
        task_id='warm_up_warehouse',
        conn_id=SNOWFLAKE_CONN,
        sql="SELECT CURRENT_TIMESTAMP();",
    )
    
    warm_up_warehouse >> check_raw_data >> run_dbt_project >> ...

    Costs almost nothing. Saves 5-10 seconds of variability at the start of every run.


    Why I Think This Is the Right Direction

    I started exploring this because nobody told me to. My team’s existing setup worked. A reasonable person would have left it alone.

    But the more I looked at this setup, the more I kept thinking about the overhead we carry when tools don’t talk to each other natively. Every boundary between systems is a place where credentials leak, latency is added, and debugging gets harder. The native dbt project in Snowflake closes one of those boundaries. Airflow still owns orchestration — which is where it belongs — but the transformation execution lives where the data lives.

    For the growing number of organizations that have standardized on Snowflake, the native integration offers something genuinely compelling: one fewer system to run, one fewer vendor to manage, and one fewer boundary between your data and the logic that transforms it.

    That sentence landed for me when I read it. That’s exactly what this is.

    If you’ve been running dbt Core on a server and Airflow alongside it and you’ve been tolerating that overhead long enough that you’ve stopped noticing it — try this weekend rebuild. You might be surprised how much lighter the pipeline feels on the other side.

    And if you do try it and hit something weird, drop it in the comments. I’m still learning this myself.

  • Automated ETL with Airflow and Python: A Practical Guide

    Automated ETL with Airflow and Python: A Practical Guide

    In the world of data, consistency is king. Manually running scripts to fetch and process data is not just tedious; it’s prone to errors, delays, and gaps in your analytics. To build a reliable data-driven culture, you need automation. This is where building an automated ETL with Airflow and Python becomes a data engineer’s most valuable skill.

    Apache Airflow is the industry-standard open-source platform for orchestrating complex data workflows. When combined with the power and flexibility of Python for data manipulation, you can create robust, scheduled, and maintainable pipelines that feed your analytics platforms with fresh data, day in and day out.

    This guide will walk you through a practical example: building an Airflow DAG that automatically fetches cryptocurrency data from a public API, processes it with Python, and prepares it for analysis.

    The Architecture: A Simple, Powerful Workflow

    Our automated pipeline will consist of a few key components, orchestrated entirely by Airflow. The goal is to create a DAG (Directed Acyclic Graph) that defines the sequence of tasks required to get data from our source to its destination.

    Here’s the high-level architecture of our ETL pipeline:

    Public API: Our data source. We’ll use the free CoinGecko API to fetch the latest cryptocurrency prices.

    Python Script: The core of our transformation logic. We’ll use the requests library to call the API and pandas to process the JSON response into a clean, tabular format.

    Apache Airflow: The orchestrator. We will define a DAG that runs on a schedule (e.g., daily), executes our Python script, and handles logging, retries, and alerting.

    Data Warehouse/Lake: The destination. The processed data will be saved as a CSV, which in a real-world scenario would be loaded into a data warehouse like Snowflake, BigQuery, or a data lake like Amazon S3.

    Let’s get into the code.

    Step 1: The Python ETL Script

    First, we need a Python script that handles the logic of fetching and processing the data. This script will be called by our Airflow DAG. We’ll use a PythonVirtualenvOperator in Airflow, which means our script can have its own dependencies.

    Create a file named get_crypto_prices.py in your Airflow project’s /include directory.

    /include/get_crypto_prices.py

    Python

    import requests
    import pandas as pd
    from datetime import datetime
    
    def fetch_and_process_crypto_data():
        """
        Fetches cryptocurrency data from the CoinGecko API and processes it.
        """
        print("Fetching data from CoinGecko API...")
        url = "https://api.coingecko.com/api/v3/simple/price"
        params = {
            'ids': 'bitcoin,ethereum,ripple,cardano,solana',
            'vs_currencies': 'usd',
            'include_market_cap': 'true',
            'include_24hr_vol': 'true',
            'include_24hr_change': 'true'
        }
        
        try:
            response = requests.get(url, params=params)
            response.raise_for_status()  # Raise an exception for bad status codes
            data = response.json()
            print("Data fetched successfully.")
    
            # Process the JSON data into a list of dictionaries
            processed_data = []
            for coin, details in data.items():
                processed_data.append({
                    'coin': coin,
                    'price_usd': details.get('usd'),
                    'market_cap_usd': details.get('usd_market_cap'),
                    'volume_24h_usd': details.get('usd_24h_vol'),
                    'change_24h_percent': details.get('usd_24h_change'),
                    'timestamp': datetime.now().isoformat()
                })
    
            # Create a pandas DataFrame
            df = pd.DataFrame(processed_data)
            
            # In a real pipeline, you'd load this to a database.
            # For this example, we'll save it to a CSV in the local filesystem.
            output_path = '/tmp/crypto_prices.csv'
            df.to_csv(output_path, index=False)
            print(f"Data processed and saved to {output_path}")
            
        except requests.exceptions.RequestException as e:
            print(f"Error fetching data from API: {e}")
            raise
    
    if __name__ == "__main__":
        fetch_and_process_crypto_data()
    

    Step 2: Creating the Airflow DAG

    Now, let’s create the Airflow DAG that will schedule and run this script. This file will live in your Airflow dags/ folder.

    We’ll use the @task decorator and the PythonVirtualenvOperator to create a clean, isolated task.

    dags/crypto_etl_dag.py

    Python

    from __future__ import annotations
    
    import pendulum
    
    from airflow.models.dag import DAG
    from airflow.operators.python import PythonVirtualenvOperator
    
    with DAG(
        dag_id="crypto_price_etl_pipeline",
        start_date=pendulum.datetime(2025, 9, 27, tz="UTC"),
        schedule="0 8 * * *",  # Run daily at 8:00 AM UTC
        catchup=False,
        tags=["api", "python", "etl"],
        doc_md="""
        ## Cryptocurrency Price ETL Pipeline
        This DAG fetches the latest crypto prices from the CoinGecko API,
        processes the data with Python, and saves it as a CSV.
        """,
    ) as dag:
        
        run_etl_task = PythonVirtualenvOperator(
            task_id="run_python_etl_script",
            python_callable_source="""
    from include.get_crypto_prices import fetch_and_process_crypto_data
    fetch_and_process_crypto_data()
    """,
            requirements=["pandas==2.1.0", "requests==2.31.0"],
            system_site_packages=False,
        )
    
    

    This DAG is simple but powerful. Airflow will now:

    • Run this pipeline automatically every day at 8:00 AM UTC.
    • Create a temporary virtual environment and install pandas and requests for the task.
    • Execute our Python function to fetch and process the data.
    • Log the entire process, and alert you if anything fails.

    Step 3: The Analytics Payoff

    With our pipeline running automatically, we now have a consistently updated CSV file (/tmp/crypto_prices.csv on the Airflow worker). In a real-world scenario where this data is loaded into a SQL data warehouse, an analyst can now run queries to derive insights, knowing the data is always fresh.

    An analyst could now answer questions like:

    • What is the daily trend of Bitcoin’s market cap?
    • Which coin had the highest percentage change in the last 24 hours?
    • How does trading volume correlate with price changes across different coins?

    Conclusion: Build Once, Benefit Forever

    By investing a little time to build an automated ETL with Airflow and Python, you create a resilient and reliable data asset. This approach eliminates manual, error-prone work and provides your analytics team with the fresh, trustworthy data they need to make critical business decisions. This is the core of modern data engineering: building automated systems that deliver consistent value.