DEV Community

Airflow vs Dagster vs Prefect: How to Choose

Airflow, Dagster and Prefect are all open-source Python orchestrators. All three schedule pipelines, retry failed steps, keep a run history, show it in a web UI and run on Kubernetes or a cloud container service.

"Which one is best?" has no general answer. What differs is the unit you model (a task, a data asset, or a Python function call), how much infrastructure you operate, and how far your existing code and skills already lean one way.

This lesson compares them as a decision framework. It assumes you know Airflow's basics from Dag fundamentals and TaskFlow and scheduling and assets, and does not re-teach them.

Versions compared

As of October 2026: Apache Airflow 3.x (3.3.2 used here), Dagster 1.x (1.13.26) and Prefect 3.x (3.8.8). Older comparisons are often wrong now.

Airflow 3 renamed datasets to assets and added event-driven scheduling; in Prefect 3 workers are standard, replacing the agents that were deprecated during Prefect 2. Dagster stays on 1.x and its documentation now scaffolds and runs projects with the create-dagster and dg command-line tools.

The same pipeline in all three

A tiny extract, transform, load job: three order rows go in, revenue per country comes out, and it is written to SQLite with INSERT OR REPLACE, so running it twice gives the same table. In each tool the business logic is the same; only the wrapping changes.

Airflow: a Dag of tasks

import sqlite3
from airflow.sdk import dag, task

DB = "orders.db"

@dag(schedule="@daily", catchup=False)
def orders_pipeline():
    @task
    def extract() -> list[dict]:
        return [
            {"country": "IN", "amount": 120.0},
            {"country": "US", "amount": 80.0},
            {"country": "IN", "amount": 30.0},
        ]

    @task
    def transform(rows: list[dict]) -> dict[str, float]:
        totals: dict[str, float] = {}
        for r in rows:
            totals[r["country"]] = totals.get(r["country"], 0) + r["amount"]
        return totals

    @task
    def load(totals: dict[str, float]) -> None:
        with sqlite3.connect(DB) as con:
            con.execute("CREATE TABLE IF NOT EXISTS revenue (country TEXT PRIMARY KEY, total REAL)")
            con.executemany("INSERT OR REPLACE INTO revenue VALUES (?, ?)", totals.items())
        print(f"loaded {len(totals)} rows: {totals}")

    load(transform(extract()))

pipeline = orders_pipeline()

if __name__ == "__main__":
    pipeline.test()

dag.test() executes a real Dag run, so the file has to sit in the Dags folder and the metadata database must exist (airflow db migrate).

Selected lines of the output (the full log is about 50 structured lines):

loaded 2 rows: {'IN': 150.0, 'US': 80.0} ... DagRun Finished: dag_id=orders_pipeline, ... run_duration=7.226455, state=success, run_type=manual, ...

The return values travel between tasks as XComs, which is fine for a small dict and wrong for a DataFrame; see XCom, Variables and Connections.

Dagster: three software-defined assets

import sqlite3
import dagster as dg

DB = "orders.db"

@dg.asset
def raw_orders() -> list[dict]:
    return [
        {"country": "IN", "amount": 120.0},
        {"country": "US", "amount": 80.0},
        {"country": "IN", "amount": 30.0},
    ]

@dg.asset
def country_revenue(raw_orders: list[dict]) -> dict[str, float]:
    totals: dict[str, float] = {}
    for r in raw_orders:
        totals[r["country"]] = totals.get(r["country"], 0) + r["amount"]
    return totals

@dg.asset
def revenue_table(country_revenue: dict[str, float]) -> dg.MaterializeResult:
    with sqlite3.connect(DB) as con:
        con.execute("CREATE TABLE IF NOT EXISTS revenue (country TEXT PRIMARY KEY, total REAL)")
        con.executemany("INSERT OR REPLACE INTO revenue VALUES (?, ?)", country_revenue.items())
    return dg.MaterializeResult(metadata={"rows": len(country_revenue)})

if __name__ == "__main__":
    result = dg.materialize([raw_orders, country_revenue, revenue_table])
    print("success: ", result.success)
    print("country_revenue: ", result.output_for_node("country_revenue"))

There is no explicit dependency wiring: the parameter name raw_orders is the dependency on the asset of that name.

Selected output lines:

... raw_orders - Writing file at: /tmp/tmpd69noogr/storage/raw_orders using PickledObjectFilesystemIOManager...
... raw_orders - ASSET_MATERIALIZATION - Materialized value raw_orders.
... revenue_table - STEP_SUCCESS - Finished execution of step "revenue_table" in 252ms.
... RUN_SUCCESS - Finished execution of run for "__ephemeral_asset_job__".
success: True
country_revenue: {'IN': 150.0, 'US': 80.0}

Note the line about the IO manager: Dagster stored each returned value (pickled, in a temporary directory) and loaded it for the downstream asset. In production you choose an IO manager that writes to your warehouse or object store, or you write the data yourself, as revenue_table does, and return only metadata.

Prefect: a flow calling tasks

import sqlite3
from prefect import flow, task

DB = "orders.db"

@task
def extract() -> list[dict]:
    return [
        {"country": "IN", "amount": 120.0},
        {"country": "US", "amount": 80.0},
        {"country": "IN", "amount": 30.0},
    ]

@task
def transform(rows: list[dict]) -> dict[str, float]:
    totals: dict[str, float] = {}
    for r in rows:
        totals[r["country"]] = totals.get(r["country"], 0) + r["amount"]
    return totals

@task
def load(totals: dict[str, float]) -> int:
    with sqlite3.connect(DB) as con:
        con.execute("CREATE TABLE IF NOT EXISTS revenue (country TEXT PRIMARY KEY, total REAL)")
        con.executemany("INSERT OR REPLACE INTO revenue VALUES (?, ?)", totals.items())
    return len(totals)

@flow(log_prints=True)
def orders_pipeline() -> int:
    rows = load(transform(extract()))
    print(f"loaded {rows} rows")
    return rows

if __name__ == "__main__":
    orders_pipeline()

Full output:

07:22:44.864 | INFO| prefect - Starting temporary server on http://127.0.0.1:8282 See https://docs.prefect.io/v3/concepts/server#how-to-guides for more information on running a dedicated Prefect server.
07:22:51.455 | INFO| Flow run 'fierce-kagu' - Beginning flow run 'fierce-kagu' for flow 'orders-pipeline'
07:22:51.533 | INFO| Task run 'extract-d55' - Finished in state Completed()
07:22:51.538 | INFO| Task run 'transform-5d2' - Finished in state Completed()
07:22:51.549 | INFO| Task run 'load-185' - Finished in state Completed()
07:22:51.550 | INFO| Flow run 'fierce-kagu' - loaded 2 rows
07:22:52.532 | INFO| Flow run 'fierce-kagu' - Finished in state Completed()
07:22:52.563 | INFO| prefect - Stopping temporary server on http://127.0.0.1:8282

Calling the flow is just calling a Python function. With no server configured, Prefect started a temporary local API server for the run and stopped it afterwards. Data passed between tasks as ordinary return values in one process.

What the three snippets show: the Airflow and Prefect versions look almost identical, but Airflow parses the file into a Dag the scheduler stores and runs task by task, while Prefect discovers the graph as the function executes. Dagster's version names things that exist (raw_orders, country_revenue, revenue_table) instead of steps that run.

Programming model

Airflow: Dags and tasks. A Dag declares tasks and their dependencies; the Dag processor parses the file and stores a serialized version, and the scheduler creates runs of it. The graph is fixed when the file is parsed, with dynamic task mapping for fan-out at run time.

Since Airflow 3, assets are first-class: a task can declare outlets, a Dag can be scheduled on assets, and the @asset decorator is shorthand for a Dag with one task that updates one asset. Assets are still attached to tasks; the task is the primary unit.

Dagster: software-defined assets. The Dagster docs define an asset as an object in persistent storage (a table, file or model) and an asset definition as code describing what should exist and how to produce it. Dependencies are declared on the assets, so the asset graph exists before anything runs, and "run this pipeline" becomes "materialize this selection of assets". Dagster also has ops, graphs and jobs for task-style work, but you do not need them to use assets.

Prefect: flows and tasks. Flows and tasks are decorated Python functions, and dependencies are whatever your code does: function calls, loops, if statements. Calling a task runs it in the caller's thread by default; .submit() and .map() send it to a task runner for concurrency (a thread pool by default, process pools, or Dask and Ray through extras). The graph is recorded at run time rather than declared up front, which makes data-dependent branching natural and makes "what will this run do?" harder to answer before it runs.

Scheduling and event triggers

  • Time-based
    • Airflow 3: Cron and presets, timetables, data intervals with catchup and backfill
    • Dagster 1.x: Schedules (cron) on jobs or asset selections; time-partitioned assets
    • Prefect 3: Schedules (cron, interval, RRule) on deployments
  • Data-aware
    • Airflow 3: Dags scheduled on asset updates, combined with AND and OR conditions
    • Dagster 1.x: Declarative Automation: conditions on assets such as on_cron, eager, on_missing
    • Prefect 3: Deployment triggers: run a deployment when matching events arrive, for example another flow run completing
  • External events
    • Airflow 3: AssetWatcher on an asset with an event trigger, for example on a message queue (added in 3.0)
    • Dagster 1.x: Sensors (polling with cursors and run keys); pushing through the GraphQL API
    • Prefect 3: Automations on events; webhooks in Prefect Cloud
  • Absence of events
    • Airflow 3: Deadline Alerts on Dag runs (experimental since 3.1)
    • Dagster 1.x: Freshness policies (still under active development) with freshness alerts
    • Prefect 3: Proactive triggers: fire when an expected event does not happen

Two details catch people out. Airflow's event-driven scheduling only accepts triggers that inherit from BaseEventTrigger, not every deferrable trigger. Dagster's Declarative Automation only acts once the default automation condition sensor is switched on in the UI. For Airflow's interval semantics (logical date, data interval, catchup), see scheduling and assets.

Data awareness and lineage

This is where the tools differ most. Dagster puts the asset graph at the centre of the UI: each asset shows its upstreams, last materialization, metadata and asset checks (data-quality tests attached to an asset). The dbt integration loads each dbt model as an asset, so dbt models and Python assets appear in one graph and can be selected together. Dagster+ adds an asset catalog with column-level lineage.

Airflow tracks assets by URI and shows which Dags produce and consume them; asset events can carry extra metadata (for example a row count). For lineage across systems, the OpenLineage provider emits lineage events to an OpenLineage-compatible backend.

Prefect models assets with the @materialize decorator and a URI key, infers dependencies from how materializations feed each other, and tracks them across workflows. Its documentation marks assets as a Prefect Cloud feature, so check availability before planning around it on a self-hosted server.

If lineage and "is this table fresh and tested?" are the questions your stakeholders ask most, that favours Dagster's model. If they ask "did last night's run succeed?", all three answer it.

Local development and testing

Airflow: dag.test() runs a whole Dag in one process for debugging, but it needs an initialised metadata database and a Dag inside a Dag bundle (the Dags folder by default). Unit tests usually call the task's Python function or load the DagBag to check that files import. The full approach is in testing and CI/CD.

Dagster: assets can be called directly in unit tests with plain inputs and mocked resources, and materialize() runs a selection in process, as in the example. dg dev starts the local UI on a project scaffolded with create-dagster.

Prefect: flows and tasks are called like functions; prefect_test_harness() runs them against a temporary database in tests. The example ran with nothing installed beyond pip install prefect.

Prefect and Dagster have the shorter loop from "edit" to "see it run". Airflow's loop got shorter with dag.test(), but you still carry a metadata database and the parse-then-run model, which matters for tests that build Dags in memory.

Deployment and scaling

  • Long-running services
    • Airflow 3: Scheduler, Dag processor, API server, metadata database; triggerer for deferrable tasks; workers depending on executor
    • Dagster OSS: Webserver, daemon (schedules, sensors, run queue), code location servers, database
    • Prefect 3: Prefect server (API, UI) with its database; workers for hybrid work pools
  • How work is placed
    • Airflow 3: Executor in the scheduler: Local, Celery, Kubernetes, Edge, AWS ECS and others; several executors at once since 2.10
    • Dagster OSS: Run launcher starts a run worker (process, Docker, Kubernetes job, ECS task); an executor inside the run runs steps (multiprocess by default, or a pod or task per step)
    • Prefect 3: Work pool holds infrastructure config; a typed worker polls it and starts each flow run (process, Docker, Kubernetes, ECS, Cloud Run and others)
  • Granularity of isolation
    • Airflow 3: Per task
    • Dagster OSS: Per run, or per step with a step executor
    • Prefect 3: Per flow run; tasks run inside it
  • User code isolation
    • Airflow 3: Task SDK: tasks talk to the API server, not the metadata database
    • Dagster OSS: Code locations run in their own processes, separate from the webserver and daemon
    • Prefect 3: Flow code is pulled from an image, Git or storage by the worker

The isolation row matters for cost and reliability. Airflow isolates tasks (with the KubernetesExecutor each task is its own pod); Prefect isolates flow runs, and tasks inside one run share its process and memory; Dagster defaults to a process per step inside a run and can go to a pod per step.

In all three, heavy computation should still run in Spark, the warehouse or a container the task launches, not in the orchestrator's workers.

Scaling the Airflow side is covered in executors,

Read on DEV Community ↗ ← Back to News

Comments

No comments yet. Start the discussion.