Dagster, Prefect, Kestra and cron: how they compare

Three orchestrators run the same canteen pipeline in this lesson, in about 35 lines of Python each: Apache Airflow 3.3.2, Dagster 1.13.26 and Prefect 3.8.8. In the last-month counts the first example reads, their packages had 7.2, 7.9 and 6.9 million PyPI downloads, and they and a fourth orchestrator, Kestra 2.0.7, all carry the Apache License 2.0.

Dagster models a pipeline as the data it keeps up to date, while Airflow, Prefect and Kestra model the steps that build that data. Airflow and Prefect write those steps in Python; Kestra writes them in YAML for its own Java server. Cron, the baseline, only starts each command on time. The shared pipeline is the nightly job of Campus Bite, this track’s invented food-ordering platform: read raw orders, drop the cancelled ones and sum revenue per canteen.

Task-centric tools think in steps, asset-centric tools in data

Two mental models split the orchestratorsA tool that runs a graph of steps on a schedule, in order, with retries, backfills and a record of each run. of this lesson. In a task-centric orchestratortask-centric orchestrator: An orchestrator that models a pipeline as a graph of steps and the order they run in, with schedules and retries set on that graph and its steps. Apache Airflow, Prefect and Kestra work this way. the unit of thought is a step: a function, a shell command, a SQL query. The pipeline is a graph of steps and their order, with its schedule and retries set on that graph and its steps. Airflow, Prefect and Kestra sit on this side. Cron is its bare end: it knows only when each command starts, and leaves job order, retries and reruns of past days to the scripts (Orchestration concepts). In an asset-centric orchestratorasset-centric orchestrator: An orchestrator that models a pipeline as a graph of the data products it keeps up to date, called assets, and derives the steps to run from that graph. Dagster is built this way. the unit is the data each step produces. The pipeline is a graph of data products, called assets, and the steps that build them follow from that graph; schedules, sensors and automation conditions decide when an asset is brought up to date (Dagster: automation). Dagster was built on this model, and Airflow 3’s assets bring part of it to a task graph. Both models can express the same pipeline, but what you monitor, test and reason about differs.

The figure below scores cron and the four orchestrators on five criteria; Kestra’s cells come from its documentation, since the lesson does not run it. Answer the three questions about your team and the figure rules out what does not fit, with the reason:

The licence column never settles it; the pipeline count, the unit of thought and who runs the servers do. Kestra 2.0.7 sits on the task-centric side but defines its flows in YAML rather than Python (Kestra: flows), and brings its own scheduler and UI in a server written in Java; its open-source edition carries the Apache License 2.0 (Kestra on GitHub). Dagster+ and Prefect Cloud are hosted services of two of these engines; this lesson stays on the open-source cores.

What the download numbers can and cannot say

The counts of the opening are PyPI Stats’ last-month figures, and the versions and licences come from PyPI’s JSON API, Kestra’s from its GitHub release and repository. Both sit in two small files beside this lesson, so the script below prints the same table on any machine:

"""Print each orchestrator's saved version, licence and last-month PyPI downloads.

The script reads two JSON files saved beside it, so the table is the same on
every run and needs no network:

- inputs/pypi_releases.json: each package's version and licence, from
  pypi.org/pypi/<name>/json (info.version, and info.license_expression or
  info.license). Kestra's server is a Java application, not a PyPI package:
  its version is its newest stable release on github.com/kestra-io/kestra
  and its licence is the repository's. (The PyPI package named kestra is
  Kestra's Python client, so its downloads would not count the server.)
- inputs/pypi_recent.json: data.last_month from
  pypistats.org/api/packages/<name>/recent, the package's downloads in the
  last month.
"""

from __future__ import annotations

import json
from pathlib import Path

HERE = Path(__file__).parent
RELEASES = json.loads((HERE / "inputs" / "pypi_releases.json").read_text())
RECENT = json.loads((HERE / "inputs" / "pypi_recent.json").read_text())

NAMES = ["apache-airflow", "dagster", "prefect", "kestra"]


def downloads(name: str) -> str:
    if name not in RECENT:
        return "server not on PyPI"
    return f"{RECENT[name]:,}"


def main() -> None:
    rows = [("Package", "Version", "Last-month downloads", "Licence")]
    for name in NAMES:
        release = RELEASES[name]
        rows.append((name, release["version"], downloads(name), release["licence"]))

    widths = [max(len(row[i]) for row in rows) for i in range(len(rows[0]))]
    print("| " + " | ".join(cell.ljust(w) for cell, w in zip(rows[0], widths)) + " |")
    print("|" + "|".join("-" * (w + 2) for w in widths) + "|")
    for row in rows[1:]:
        print("| " + " | ".join(cell.ljust(w) for cell, w in zip(row, widths)) + " |")

    licences = sorted({RELEASES[name]["licence"] for name in NAMES})
    print()
    if len(licences) == 1:
        print(f"All {len(NAMES)} carry one licence: {licences[0]}")
    else:
        print("Licences: " + ", ".join(licences))


if __name__ == "__main__":
    main()

| Package        | Version | Last-month downloads | Licence    |
|----------------|---------|----------------------|------------|
| apache-airflow | 3.3.2   | 7,196,244            | Apache-2.0 |
| dagster        | 1.13.26 | 7,886,605            | Apache-2.0 |
| prefect        | 3.8.8   | 6,895,663            | Apache-2.0 |
| kestra         | 2.0.7   | server not on PyPI   | Apache-2.0 |

All 4 carry one licence: Apache-2.0

In that month the three Python packages were within about one million downloads of each other. Kestra’s row has no count because its server is a Java application released on GitHub; the PyPI package named kestra is its Python client (PyPI: kestra). The Python Packaging User Guide adds that pip’s cache, mirrors and inflation scripts skew such counts, and that many downloads do not make a project good (Python Packaging User Guide). Close counts say only that all three are widely installed.

Practise

The exercise holds the canteen job twice, as written-down steps and as asset functions that only name what they read: derive the run order from those reads.

"""Starter for the asset-run-order exercise.

The Campus Bite nightly job written two ways, on the same five orders.

TASKS is the task-centric view: three steps in the order someone wrote down,
each with the data product it builds. ASSETS is the asset-centric view: each
product and the function that builds it, and each function names the
products it reads as its parameters. Change run_order() so it derives the
order from those reads, and materialise() so it builds every product once,
in that order.
"""

from __future__ import annotations

import inspect

RAW = [
    {"canteen": "North Block", "amount_inr": 90, "status": "delivered"},
    {"canteen": "North Block", "amount_inr": 60, "status": "delivered"},
    {"canteen": "North Block", "amount_inr": 55, "status": "cancelled"},
    {"canteen": "Hostel 3", "amount_inr": 80, "status": "delivered"},
    {"canteen": "Hostel 3", "amount_inr": 70, "status": "delivered"},
]

# The task-centric view: each step and the product it builds, in the order written by hand.
TASKS = [
    ("extract", "raw_orders"),
    ("transform", "delivered_totals"),
    ("load", "canteen_revenue"),
]


def raw_orders() -> list[dict]:
    return list(RAW)


def delivered_totals(raw_orders: list[dict]) -> dict[str, int]:
    totals: dict[str, int] = {}
    for row in raw_orders:
        if row["status"] == "delivered":
            totals[row["canteen"]] = totals.get(row["canteen"], 0) + row["amount_inr"]
    return totals


def canteen_revenue(delivered_totals: dict[str, int]) -> list[dict]:
    return [{"canteen": c, "revenue_inr": t} for c, t in sorted(delivered_totals.items())]


# The asset-centric view: each product and the function that builds it, in no particular order.
ASSETS = {
    "canteen_revenue": canteen_revenue,
    "raw_orders": raw_orders,
    "delivered_totals": delivered_totals,
}


def reads(build) -> list[str]:
    """The products a build function reads: its parameter names."""
    return list(inspect.signature(build).parameters)


def run_order(assets: dict) -> list[str]:
    """Every product name once, each one after every product it reads."""
    return list(assets)


def materialise(assets: dict) -> dict:
    """Build every product once, in run order, passing each the values it reads."""
    values: dict = {}
    for name, build in assets.items():
        values[name] = build(*[values[upstream] for upstream in reads(build)])
    return values

"""Sample tests for the asset-run-order exercise.

The derived order must match the task view's written order, the products must
come out right, new products must find their own places from what they read,
and a circle of reads must be refused.
"""

from __future__ import annotations

from canteen_views import ASSETS, TASKS, materialise, reads, run_order


def placed_after_their_reads(order: list[str], assets: dict) -> bool:
    return all(order.index(upstream) < order.index(name) for name in assets for upstream in reads(assets[name]))


def test_the_derived_order_matches_the_written_one():
    """The asset view derives the order the task view wrote by hand."""
    written = [product for _step, product in TASKS]
    derived = run_order(ASSETS)
    assert derived == written, f"the reads give {written}, not {derived}"


def test_materialise_builds_the_revenue_rows():
    """Each canteen's delivered orders add up to 150 rupees; the cancelled 55 is left out."""
    values = materialise(ASSETS)
    assert values["delivered_totals"] == {"North Block": 150, "Hostel 3": 150}, (
        f"delivered_totals is {values.get('delivered_totals')}: only delivered orders count"
    )
    assert values["canteen_revenue"] == [
        {"canteen": "Hostel 3", "revenue_inr": 150},
        {"canteen": "North Block", "revenue_inr": 150},
    ], f"canteen_revenue is {values.get('canteen_revenue')}"
    assert sorted(values) == sorted(ASSETS), "materialise must return one value for every product"


def test_new_products_find_their_own_places():
    """Two new products join without anyone writing an edge; each lands after what it reads."""

    def top_canteen(canteen_revenue: list[dict]) -> str:
        return max(canteen_revenue, key=lambda row: (row["revenue_inr"], row["canteen"]))["canteen"]

    def order_count(raw_orders: list[dict]) -> int:
        return len(raw_orders)

    assets = {"top_canteen": top_canteen, **ASSETS, "order_count": order_count}
    order = run_order(assets)
    assert sorted(order) == sorted(assets), f"run_order must list each of the 5 products once, not {order}"
    assert placed_after_their_reads(order, assets), f"in {order} a product comes before something it reads"
    assert order == run_order(dict(reversed(list(assets.items())))), (
        "the order must not depend on the order of the dictionary: visit the names sorted"
    )
    values = materialise(assets)
    assert values["order_count"] == 5, f"order_count is {values.get('order_count')}, not the 5 raw orders"
    assert values["top_canteen"] == "North Block", f"top_canteen is {values.get('top_canteen')}"


def test_each_product_is_built_once():
    """A product read by two others is still built once."""
    calls: list[str] = []

    def base() -> int:
        calls.append("base")
        return 2

    def double(base: int) -> int:
        calls.append("double")
        return base * 2

    def total(base: int, double: int) -> int:
        calls.append("total")
        return base + double

    values = materialise({"total": total, "double": double, "base": base})
    assert values == {"base": 2, "double": 4, "total": 6}, f"the values are {values}"
    assert sorted(calls) == ["base", "double", "total"], f"the builds were {calls}: each product once"


def test_a_circle_of_reads_is_refused():
    """Two products that read each other have no run order."""

    def first(second: int) -> int:
        return second

    def second(first: int) -> int:
        return first

    try:
        run_order({"first": first, "second": second})
    except ValueError:
        return
    raise AssertionError("a circle of reads must raise ValueError, not return an order")

Six questions then check the two models, the figure and the download numbers.

The same canteen pipeline, in Airflow, Dagster and Prefect Pro: opens with a Pro, Premium or Ultimate pass

Deployment, testing and local development compared Pro: opens with a Pro, Premium or Ultimate pass

Interview questions

Warm-up (fresher to 2-5 years): What makes four orchestrators the same, and where do they differ?

All four run a graph of work on a schedule, in order, with retries you can set and a record of every run; cron only starts each command on time. They differ in the unit of thought: Airflow, Prefect and Kestra model the steps and their order, while Dagster models the data products and derives the steps from them. They also differ in how a pipeline is written: in Python for three of them, and in YAML for Kestra, which runs it on its own Java server.

Answer each prompt below before you open its first step.

Key takeaways

• Task-centric tools (Airflow, Prefect, Kestra) model a pipeline as a graph of steps and cron only starts each command; asset-centric tools (Dagster, and Airflow’s assets) model the data products and derive the steps from them.
• All four orchestrators carry the Apache License 2.0, and in the saved month the three Python packages were within about one million PyPI downloads of each other, so neither the licence nor the counts pick the tool.
• In the figure, cron plus a tested script fits one or two jobs and drops out past ten pipelines; a team that cannot run servers needs a hosted service, and a team that reasons about data leans towards Dagster or Airflow’s assets.