Migrating Media DAGs from Airflow to Prefect

This page sits beneath Choosing a Media Pipeline Orchestrator within the Pipeline Automation & Batch Processing layer. It assumes you have already made the decision — a media pipeline’s graph shape is dynamic enough, and its scheduler footprint heavy enough, that Prefect earns the switch — and now you need the concrete mechanics of moving working Airflow DAGs to Prefect flows without a big-bang rewrite or a lost weekend of re-encodes.

Problem Framing

An Airflow-to-Prefect migration is not a syntax translation. The two systems disagree on the things a media pipeline leans on hardest: how state passes between steps, how dynamic fan-out is expressed, and where large artifacts live. The migration fails in predictable ways when teams treat it as a find-and-replace. The most damaging symptom shows up the first time a transcode result flows through the new Prefect flow:

prefect.exceptions.SerializationError: result for task 'encode-1080p' is 2.14 GiB;
  exceeded result store inline limit — refusing to persist rendition bytes

That error is the same mistake that quietly worked (badly) on Airflow: XCom will happily pickle a large payload into the metadata database and rot it from the inside, while Prefect’s result store refuses outright. Both are telling you the same thing — media bytes must never travel through the orchestrator’s state channel. The migration is the moment to fix the handoff, not port the bug. The other two friction points are dynamic mapping (Airflow’s .expand() versus Prefect’s .map(), which differ in join semantics) and retries (per-operator config versus per-task decorator with result caching).

Solution Architecture

Approach the migration as a concept map, not a rewrite. Every Airflow primitive your media DAG uses has a Prefect equivalent; translate them one row at a time and the flow falls out. The diagram is that mapping, with the XCom-to-result row highlighted because it is the one that changes behavior rather than just syntax.

The most important thing the map communicates is that most rows are a straight translation — only the XCom row demands a behavioral change, because Prefect refuses to carry the media bytes that Airflow tolerated. Fix that one row correctly and the rest is mechanical.

Implementation

1. Translate the DAG to a flow with mapped tasks

Start with a representative Airflow media DAG and its Prefect equivalent side by side. The Airflow original, a dynamically-mapped transcode:

# apache-airflow==2.9.3  (BEFORE)
from airflow.decorators import dag, task
import pendulum

@dag(schedule="0 3 * * *", start_date=pendulum.datetime(2025, 5, 1), catchup=False)
def transcode_dag():
    @task
    def list_jobs() -> list[dict]:
        return discover_targets()

    @task(retries=3, retry_delay=pendulum.duration(seconds=30))
    def encode(job: dict) -> dict:
        # BAD on Airflow, fatal on Prefect: returning bytes-like state
        return {"uri": run_ffmpeg(job["src"], job["profile"]), **job}

    @task
    def package(rends: list[dict]):
        return build_hls_master(rends, "nightly")

    package(encode.expand(job=list_jobs()))

transcode_dag()

The Prefect flow keeps the shape but expresses fan-out as .map() and control flow as plain Python. Consult the official Prefect flows and tasks documentation for the current decorator signatures.

# prefect==3.1.5  (AFTER)
from prefect import flow, task

@task
def list_jobs() -> list[dict]:
    return discover_targets()

@task(retries=3, retry_delay_seconds=30, persist_result=True, timeout_seconds=2700)
def encode(job: dict) -> dict:
    # store ONLY the reference: the URI + digest, never the rendition bytes
    return {"uri": run_ffmpeg(job["src"], job["profile"]),
            "sha256": digest_of(job), **job}

@task
def package(rends: list[dict]) -> str:
    return build_hls_master(rends, "nightly")

@flow(name="nightly-transcode")
def transcode_flow():
    jobs = list_jobs()
    futures = encode.map(jobs)               # runtime fan-out, like .expand()
    return package([f.result() for f in futures])   # explicit fan-in join

The join is the subtle difference. Airflow’s package(encode.expand(...)) infers the gather from the dependency; in Prefect you gather explicitly with [f.result() for f in futures], which is more code but removes the ambiguity about when the join resolves — it blocks until every mapped run completes.

2. Replace XCom with results stored by reference

This is the load-bearing change. In Airflow, a task returning a dict lands in XCom in the metadata database. In Prefect, a returned value becomes a Result — and if you configure a result store, Prefect persists it. Neither should ever carry the media payload. Configure the result store to hold the small reference record and keep the gigabytes in object storage, exactly the handoff: reference contract the parent decision framework mandates.

# prefect==3.1.5
from prefect.filesystems import RemoteFileSystem
from prefect import flow

# Result store holds the small JSON record (uri + sha256), NOT the bytes.
result_store = RemoteFileSystem(basepath="s3://mp-prefect-results")

@flow(result_storage=result_store, persist_result=True)
def transcode_flow():
    ...

With this in place, a re-run of a partially-failed nightly job reads the persisted references and skips already-encoded renditions via result caching — the Prefect analogue of an Airflow rerun skipping successful task instances, but keyed on task inputs rather than DAG-run state. The rule to enforce in code review: any task whose return value could exceed a few kilobytes returns a URI and a digest, never the object itself.

3. Map pools and retries

Pool and max_active_tasks become work-pool concurrency limits; retries and retry_delay become retries and retry_delay_seconds on the task decorator (already shown above). Set the GPU backpressure limit on the work pool the encode tasks run in:

prefect work-pool create gpu-pool --type process
prefect work-pool set-concurrency-limit gpu-pool 2   # = GPU count, replaces the Airflow pool

Verification

Cut over one pipeline at a time — never all DAGs at once — and reconcile against the Airflow history before decommissioning anything.

# 1. Prove the result store carries references, not payloads: records must be tiny.
prefect deployment run 'nightly-transcode/prod'
aws s3 ls --recursive s3://mp-prefect-results/ | awk '{ if ($3 > 100000) print "FAT RESULT:", $4 }'
# expect: no output (every result record well under 100 KB)

# 2. Confirm dynamic mapping fanned out to the same job count as the old DAG.
prefect flow-run logs <run-id> | grep -c "Created task run 'encode-"

# 3. Confirm re-run caching skips completed encodes (idempotent recovery).
prefect deployment run 'nightly-transcode/prod'
prefect flow-run logs <run-id> | grep -c "Finished in state Cached"   # >0 on the second run

# 4. Confirm GPU backpressure: never more than 2 encode runs concurrently.
prefect work-pool inspect gpu-pool | grep concurrency_limit          # 2

Run the Prefect flow in shadow alongside the live Airflow DAG for a few cycles, diff the produced renditions by digest, and only pause the Airflow DAG once the digests match on every profile. That parallel-run window is what makes the cutover incremental rather than a leap.

Failure Modes & Edge Cases

FAQ

Related

• Up to the decision framework: Choosing a Media Pipeline Orchestrator
• Sibling head-to-head: Celery vs Airflow for Video Encoding Queues
• The Airflow starting point: orchestrating pipelines with Airflow
• Safe retries across the cutover: retry logic and dead-letter queues
• Pipeline overview: Pipeline Automation & Batch Processing