Migrating from Airflow to Prefect

Fifteen posts taught you to read Prefect with an Airflow eye. This one is about moving — taking a deployment that already runs your business on Airflow and getting it onto Prefect without a weekend of downtime and a rollback plan you never get to test. Migration is a different skill from translation. You know by now that a DAG maps to a flow; the question that decides whether the project succeeds is which DAG you port first, what runs both stacks the night you cut over, and which three Airflow features have no clean Prefect answer so you stop looking for one.

So this is the working finale: a concept map with the port baked into each row, a strategy for running both engines side by side instead of a big-bang rewrite, and an honest list of where the translation leaks. Nothing here assumes you’ll like the answer — some pipelines should stay on Airflow, and the last section names which.

The concept map, as a port

The first post gave you the noun-for-noun table for understanding Prefect. This is the version you keep open while you’re actually rewriting a file — every row is a thing you’ll delete from the Airflow side and the shape it takes on the Prefect side.

Airflow	Prefect	How you port it
`with DAG(...)` block	`@flow` function	Delete the context manager; the function body is the graph.
`PythonOperator(python_callable=fn)`	`@task def fn`	Decorate the callable; call it directly instead of wrapping it.
`BashOperator(bash_command=...)`	`@task` calling `subprocess.run`	The shell command moves into a task body; you own the invocation.
`xcom_push` / `xcom_pull`	Return value / argument	Return the value; pass it to the next task as a parameter.
`Connection` (`conn_id`)	Block (`.load(name)`)	Register a typed block; load it by name where the hook used to unpack.
`Variable.get(...)`	`Variable.get(...)`	Near-identical; `prefect.variables.Variable`, JSON by key.
`Pool` (slots)	Tag concurrency limit	Tag the tasks; set a named limit on the tag.
`Sensor` (poke)	Event + automation, or a polling task	Prefer push (event triggers a run); fall back to a task that polls.
`executor =` in config	Work pool + task runner	Where runs land (pool) is split from how tasks fan out (runner).
`KubernetesExecutor`	Kubernetes work pool	One `prefect work-pool create --type kubernetes`; deploy against it.
`SubDagOperator` / `TaskGroup`	Subflow	A flow calls another flow; ordinary function composition.
`schedule=` on the DAG	Schedule on the deployment	The schedule leaves the code and lives on the deployment.
`on_failure_callback`	State hook or automation	`on_failure=[hook]` in-process, or an automation on the failed event.
`airflow dags backfill`	`run_deployment` loop	You trigger one parameterized run per date; there’s no catchup window.

Most rows are the mechanical swaps you’ve already seen. The ones worth code are the ones the first post didn’t work through, because they’re where a naive port goes wrong.

BashOperator → a task that shells out. Airflow’s BashOperator is templated, retried, and logged by the framework. When you move it, you keep the retries and logging (they’re on @task) but you own the subprocess call, and you should own it deliberately — check=True so a nonzero exit becomes a Python exception the task can retry, and captured output so it lands in the flow logs.

The Airflow version:

from airflow.providers.standard.operators.bash import BashOperator

run_dbt = BashOperator(
    task_id="dbt_run",
    bash_command="cd /opt/bookshop && dbt run --select marts",
)

The Prefect version:

import subprocess
from prefect import task, get_run_logger

@task(retries=2)
def dbt_run(select: str = "marts"):
    logger = get_run_logger()
    result = subprocess.run(
        ["dbt", "run", "--select", select],
        cwd="/opt/bookshop",
        check=True,               # nonzero exit -> CalledProcessError -> a retry
        capture_output=True,
        text=True,
    )
    logger.info(result.stdout)
    return result.stdout

The command is now an argument, not a Jinja-templated string, so parameterizing it is a function parameter rather than a params dict and a {{ }}. (If you’d rather not hand-roll it, prefect-shell ships a ShellOperation block that wraps exactly this pattern — worth it once you have more than a couple.)

Pool → a tag concurrency limit. An Airflow pool caps how many tasks touch a scarce resource at once — the classic case is “at most three tasks may hit the warehouse.” In Prefect the cap rides on a tag, not a named pool object, and you set the limit once with the CLI:

prefect concurrency-limit create warehouse 3

@task(tags=["warehouse"])
def load_partition(day: str):
    ...

Every task tagged warehouse now shares three slots across all runs, exactly like the pool did — a run that can’t get a slot waits rather than piling on. The mental shift is that the limit is a property of the tag, applied wherever that tag appears, instead of a slot count you assign a task to. (For a limit around an arbitrary block of code rather than a whole task, there’s the concurrency("warehouse") context manager — same limiter, finer scope. The concurrency post has the full story.)

KubernetesExecutor → a Kubernetes work pool. This is where the executor split from post one pays off concretely. In Airflow, choosing Kubernetes means an executor = KubernetesExecutor in airflow.cfg and a pod_template_file, and it’s a deployment-wide decision. In Prefect it’s one work pool you create, and only the flows you deploy against it run as pods:

prefect work-pool create --type kubernetes bookshop-k8s

if __name__ == "__main__":
    daily_load.deploy(
        name="nightly",
        work_pool_name="bookshop-k8s",
        image="registry.example.com/bookshop:latest",
    )

Now daily_load runs as a Kubernetes job when it’s scheduled, while a different flow deployed to a process pool keeps running as a plain subprocess on a VM. You didn’t reconfigure the platform; you pointed one deployment at one pool. The work pools post covers the worker you run in-cluster to pull those jobs.

on_failure_callback → a state hook or an automation. Airflow’s on_failure_callback is a function the scheduler calls when a task fails. Prefect gives you two places to put that logic, and which one you pick is a real design choice. In-process, a state hook is a function attached to the flow that fires on a state transition:

def alert_oncall(flow, flow_run, state):
    send_slack(f"{flow_run.name} failed: {state.message}")

@flow(on_failure=[alert_oncall])
def daily_load():
    ...

That’s the closest one-to-one port, and the hook signature (flow, flow_run, state) is stable. But the more Prefect-native move is to leave the flow clean and put the reaction in an automation — a rule on the server that watches the event stream and fires when any run enters Failed, sending the notification without a line of alerting code in your flow. The hook couples the reaction to this flow; the automation decouples it and covers every flow at once. Port the callback to a hook if it’s flow-specific cleanup; port it to an automation if it’s operational alerting. The states and hooks post draws that line in detail.

backfill → a loop over run_deployment. This is the migration’s sharpest edge and it deserves its own section, but the mechanics belong in the table. Airflow’s airflow dags backfill -s 2026-01-01 -e 2026-01-31 daily_load walks a date range and materializes a run per interval, tracked as catchup. Prefect has no catchup concept — a flow doesn’t know its “logical date,” so there’s nothing to catch up on. You reproduce a backfill by triggering parameterized runs yourself:

from datetime import date, timedelta
from prefect.deployments import run_deployment

def backfill(start: date, end: date):
    day = start
    while day <= end:
        run_deployment(
            name="daily-load/nightly",
            parameters={"run_date": day.isoformat()},
            timeout=0,                      # fire-and-forget — see below
        )
        day += timedelta(days=1)

Each call triggers one real run of the deployment with that day as a parameter, and they queue and execute like any scheduled run — visible in the UI, retried, logged.

That timeout=0 is not decoration. run_deployment defaults to timeout=None, and in this API None means poll indefinitely — the call blocks until that run finishes. Leave it out and your “backfill” is a serial queue: it triggers July 1st, waits for it to complete, then triggers July 2nd. Thirty-one nightly loads, one after another, when the whole point of a backfill is that they can run concurrently. timeout=0 submits and returns immediately, and the work pool’s concurrency limit — not your while loop — decides how many run at once, which is exactly where that decision belongs.

The other catch is that your flow must take the date as a parameter rather than reading it from an execution context, because there is no execution context to read. A flow that assumed {{ ds }} was ambient has to be rewritten to accept run_date before it can be backfilled at all. That’s the single most common surprise in a real port, and it’s a design fix, not a config flag.

Sensor → an event trigger, or an honest polling task. An Airflow sensor is a task that sits in the graph and pokes — an S3KeySensor reoccupying a worker slot every thirty seconds until the file lands. Porting it well means not reproducing the poke. The native move is to flip it to push: the thing that drops the file emits an event, and an automation triggers your flow when the event arrives. If the producer is another Prefect flow, that’s emit_event("bookshop.orders.landed") on its side and an automation on yours — no polling, no burned slot. But you don’t always control the producer, and then the honest port is a task that polls, written as a plain retry loop rather than a framework sensor:

from prefect import task
from prefect_aws import S3Bucket

@task(retries=60, retry_delay_seconds=30)
def wait_for_orders(key: str):
    bucket = S3Bucket.load("bookshop-raw")
    if key not in {o["Key"] for o in bucket.list_objects()}:
        raise FileNotFoundError(key)   # a "retry" is the next poke
    return key

Sixty retries at thirty seconds is a thirty-minute poke window, and it reads as exactly what it is. The difference from an Airflow sensor is that you see the polling — it’s a retrying task, not a magic operator with a hidden poke_interval — and you can reach for the event-driven version the moment you control the producer. Prefer push; fall back to this.

TaskGroup / SubDAG → a subflow. An Airflow TaskGroup bundles tasks for visual grouping and reuse; a SubDagOperator (deprecated, but still in old code) nests a whole DAG. Both collapse into the plainest Prefect construct there is — a flow calling a flow:

@flow
def validate_orders(orders: list[dict]) -> list[dict]:
    ...  # the "task group" is just its own flow

@flow
def daily_load():
    orders = extract_orders()
    clean = validate_orders(orders)   # subflow call; runs and reports as a unit
    load(clean)

The subflow gets its own run record, its own state, and its own retries, and it composes by being called — no TaskGroup context manager, no SubDagOperator wiring, no separate DAG file. Grouping stops being an orchestration construct and becomes what it always should have been: a function.

A strategy that isn’t a rewrite weekend

The failure mode of an orchestrator migration is treating it as a rewrite: freeze features, port everything, cut over on a Saturday, discover the backfill semantics changed at 2 a.m. Don’t. Both engines are just processes that run Python and hit a database, and there is no rule that says only one may be alive. Run them side by side and migrate by pipeline.

Migrate by pipeline, not by feature. Pick one DAG and move the whole thing — flow, deployment, schedule, alerting — until it runs green on Prefect for a week. Then pick the next. The tempting alternative, “port all the PythonOperators, then all the schedules, then all the alerts,” leaves you with two half-working systems and no pipeline you can trust on either. One fully-migrated pipeline is worth ten half-migrated concepts, because it proves the whole path end to end — including the parts (deployment, worker, alerting) you can’t validate from a code diff.

Port a self-contained DAG first. The right first pipeline has no upstream or downstream dependency on another DAG, a single external system, and a schedule you can afford to run twice. A nightly extract-transform-load into one warehouse table is ideal; the DAG that fifteen other DAGs wait on via a sensor is the worst possible starting point. You want the first migration to teach you the mechanics — deploy, worker, block, schedule — without also teaching you cross-DAG coordination on day one. Save the entangled ones for after you trust the path.

Keep the schedules on Airflow until the flow is proven. Deploy the Prefect flow with its schedule paused and trigger it by hand for a few days, comparing its output against the Airflow run that’s still the source of truth. When they agree, pause the Airflow DAG and unpause the Prefect schedule in the same change. Re-pointing a schedule is two toggles, and doing them together — Airflow off, Prefect on — means there’s never a window where both write or neither does. If Prefect misbehaves, you flip it back; the Airflow DAG never left.

Bridge the two during the transition. A pipeline mid-migration often needs to talk across the fence, and both directions are a few lines. To trigger a Prefect flow from an Airflow DAG that hasn’t moved yet, drop a PythonOperator that calls run_deployment:

from airflow.providers.standard.operators.python import PythonOperator
from prefect.deployments import run_deployment

def kick_prefect():
    run_deployment(name="daily-load/nightly", timeout=0)  # fire-and-forget

trigger = PythonOperator(task_id="trigger_prefect", python_callable=kick_prefect)

timeout=0 returns immediately instead of blocking the Airflow worker until the flow finishes — the Airflow task’s job is to start the Prefect run, not babysit it. To go the other way and trigger a still-on-Airflow DAG from a Prefect flow, call Airflow’s REST API from a task:

import requests
from prefect import task
from prefect.blocks.system import Secret

@task
def trigger_airflow_dag(dag_id: str, conf: dict):
    token = Secret.load("airflow-api-token").get()
    resp = requests.post(
        f"https://airflow.example.com/api/v2/dags/{dag_id}/dagRuns",
        json={"conf": conf},
        headers={"Authorization": f"Bearer {token}"},
    )
    resp.raise_for_status()
    return resp.json()["dag_run_id"]

With those two bridges you can migrate the middle of a dependency chain: the upstream Airflow DAG triggers your new Prefect flow, which triggers the still-Airflow downstream, and nobody notices the seam. The chain migrates one link at a time, and at every step something real is running the business.

Gotchas worth knowing before you commit

The mapping table is honest about shape but quiet about friction. Four things will bite, and knowing them before the port beats discovering them during it.

There is no backfill UI like Airflow’s. This is the one that generates the most “wait, where is it” in a migration. Airflow gives you a first-class backfill: a date range, a catchup toggle, a grid view that shows exactly which intervals ran and lets you clear-and-rerun a cell. Prefect has none of that as a built-in surface. A backfill is a script you write — the run_deployment loop above — and the “which dates ran” question is answered by filtering the runs list, not by a purpose-built grid. For a team whose operational muscle memory is “clear the failed cell in the tree view and let it rerun,” this is a genuine downgrade in ergonomics, and it’s worth building a small internal backfill helper early so nobody hand-writes the loop each time. The flip side is that a backfill is just parameterized runs, so it’s scriptable, parameterizable, and testable in ways Airflow’s baked-in version isn’t — but you are trading a UI for a script, and you should say so out loud when you pitch the migration.

The provider ecosystem is smaller — map to prefect-* or plain SDKs. Airflow’s superpower is that there is already an operator for the obscure thing: a SnowflakeToS3Operator, a GoogleCloudStorageToBigQueryOperator, hundreds of them. Prefect’s integration libraries — prefect-aws, prefect-gcp, prefect-snowflake, prefect-dbt, prefect-shell, and the rest — cover the common systems with blocks and helpers, but they’re a tidy set, not an exhaustive catalog. When you hit an Airflow operator with no prefect-* equivalent, the port is a task that calls the vendor’s own SDK — boto3, google-cloud-storage, a snowflake-connector-python cursor — which is usually five lines and often clearer than the operator’s dozen parameters. But budget for it: “find the operator, fill in its kwargs” becomes “write the SDK call,” and for a DAG that leaned on ten exotic operators that’s real work, not a decoration swap.

The dynamic-by-default mindset shift is the deep one. Airflow parses your DAG file at the top level, repeatedly, to build the graph — which is why top-level code has to be cheap, why you don’t put a database query at module scope, and why dynamic shape needs a mapping API. Prefect has no parse step; the graph is discovered as the function runs, so a for loop over a query result is the fan-out and an if is the branch. The trap on the way in is porting an Airflow developer’s habits rather than their DAG: writing a Prefect flow that still avoids “expensive top-level code” that no longer exists, or reaching for a mapping helper where a plain loop would do. The port isn’t just syntactic — the whole reason to avoid dynamic shape in Airflow evaporates, and code that fought the parser can be deleted, not translated. Miss that and you’ll write timid Prefect that reads like Airflow with different imports.

Sometimes Airflow is still the right home. A migration project needs a stop rule, and the honest one is: not every DAG should move. A large, fixed, nightly batch graph feeding a warehouse — the same forty tasks in the same order every night, leaning on a dozen mature providers, with an ops team whose runbooks are built around the grid view and backfill UI — gains little from Prefect and gives up a decade of hardening and a picture of the graph you can read before it runs. Prefect earns its place on the pipelines that are dynamic: shape that depends on the data, workflows that double as background jobs, event-triggered work, config that wants to be typed objects, pipelines that change often enough that the ceremony hurts. The right output of an evaluation is a split — the dynamic, Python-heavy, fast-changing pipelines move; the big fixed batch graphs stay — not a mandate to migrate everything. A team that ports its whole estate to prove a point usually ends up wishing half of it were back on Airflow.

Final thoughts

Sixteen posts, and the arc was never “Prefect wins.” It was: here is what a pipeline looks like when the orchestrator reads order off running Python instead of asking you to declare a graph, and here is how every Airflow concept you own translates into that world — cleanly for most rows, with a real shift for a few, and not at all for the three features (transactions, automations, background tasks) Airflow simply doesn’t have.

Migration is where that translation stops being an intellectual exercise and meets your actual deployment — the DAG that pays the bills, the ops team with muscle memory, the backfill someone will need at 2 a.m. The right way through it is the unglamorous one: both engines alive at once, one pipeline moved at a time, schedules re-pointed only after a flow has proven itself for a week, and a clear-eyed decision about which pipelines shouldn’t move at all. The flashy rewrite weekend is how migrations fail; the boring pipeline-by-pipeline crawl is how they succeed.

And the deepest thing the series had to teach can’t be ported, only learned: Prefect and Airflow are two coherent answers to what a pipeline is — dynamic Python versus a scheduled graph — and the skill isn’t fluency in one, it’s knowing which of your pipelines is which. You can read a Prefect codebase now and see the Airflow shape under it. You know which rows in the map are clean swaps and which are mindset shifts. You know where Prefect’s transactions and automations give you things you were building by hand, and where Airflow’s static graph and provider catalog were never the thing slowing you down. Take that judgment to your own estate, split it honestly, and move the half that’s fighting the graph. That’s the whole job — and it’s where this series leaves you.

Examples in this chapter are docs-checked against the series’ stated versions (Prefect 3.x, Airflow 3.x), but they were not executed in this repository.

The concept map, as a port

A strategy that isn’t a rewrite weekend

Gotchas worth knowing before you commit

Final thoughts

Comments