Prefect 3.x for Data Engineering: Flows, Deployments, Work Pools & Automations

prefect is the modern Python-first orchestrator that senior data engineers reach for when Airflow's static-DAG model, per-schedule scheduler, and cron-centric worldview stop matching the workload — dynamic fan-outs, event-driven runs, sub-second reactive triggers, and lakehouse-shaped compute pools that don't fit the "one worker pool per DAG" mould. Every pipeline your team ships in 2026 has at least one of those shapes; and every mid-sized data platform eventually asks the "should we migrate off Airflow?" question. The engineering trade-off does not live in "is Prefect better" — the more useful question is which orchestration invariants Prefect 3 gets right and what migrating off Airflow actually costs, because the answer determines whether your prefect deployment config replaces a DAG-per-file or a whole 2,000-line dags/ tree.

This guide is the senior-DE walkthrough you wished existed the first time an interviewer asked "compare prefect vs airflow on dynamic DAGs and event-driven triggers," or "what changed between prefect 3 and Prefect 2 — why did they kill agents?", or "how would you design a prefect work pool layout for Kubernetes + serverless side by side?" It walks through the four canonical concepts of prefect 3 — flows and tasks (the @flow and @task decorators), deployments and work pools (versioned flow configs bound to compute abstractions), automations and events (the reactive engine that turns orchestration into a compound-condition trigger loop), and the head-to-head against Airflow 3 and Dagster — the four axes interviewers actually probe (dynamic DAG support, eventful triggers, asset lineage, migration cost), the canonical prefect.yaml layout, and the "just Python" positioning that separates Prefect from Dagster's software-defined-asset model. Each section pairs a teaching block with a Solution-Tail interview answer — code, a step-by-step trace, an output table, then a concept-by-concept breakdown of why it works.

When you want hands-on reps immediately after reading, drill the SQL practice library →, rehearse on the design practice library →, and sharpen the streaming axis with the streaming practice library →.

---

On this page

• Why Prefect 3 matters in 2026
• Flows, tasks & subflows
• Deployments + work pools
• Automations, events & triggers
• Prefect vs Airflow / Dagster + interview signals
• Cheat sheet — Prefect 3 recipes
• Frequently asked questions
• Practice on PipeCode

---

1. Why Prefect 3 matters in 2026

Four concepts, one Python-first orchestrator — the modern Airflow alternative that prioritises dynamic and eventful workflows

The one-sentence invariant: prefect 3 is a Python-first orchestrator whose four load-bearing primitives — @flow-decorated Python functions, @task-decorated subroutines returning futures, deployment objects that version a flow against a work pool compute abstraction, and automations that react to typed events with compound-condition triggers — collapse the four things Airflow makes hard (dynamic DAGs, sub-second triggers, per-flow compute, and reactive alerting) into ordinary Python plus a small YAML file, and the 2024 rewrite from Prefect 2 to Prefect 3 removed the last two operational rough edges (the "agent" process and the async-first engine) in favour of workers pulling from work pools and a sync-first execution model. Every downstream decision on a Prefect stack — how you cut deployments, how many work pools you run, whether you buy Prefect Cloud or self-host, how you migrate off Airflow — follows from those four primitives.

The four axes interviewers actually probe.

• Dynamic DAG support. Prefect flows are ordinary Python; loops, conditionals, and runtime task-mapping over an unknown-at-parse-time list all work naturally because a flow is a function call, not a parsed YAML/Python module that produces a static graph. Airflow 2/3 supports dynamic task mapping but constrains it (must return before scheduling; per-invocation limits); Prefect makes it the default. Interviewers open with this axis because it's the shape most likely to break Airflow at scale.
• Eventful triggers. Prefect 3 emits typed events on every state transition (prefect.flow-run.Completed, prefect.deployment.Created, etc.) and lets you write automations whose triggers combine events with compound predicates ("flow X completed AND flow Y not started within 5 min → notify"). Airflow's Datasets, sensors, and callbacks are the closest analogues but sit lower in the stack. Interviewers probe this axis because reactive orchestration is the shape that data-quality and SLA workflows demand.
• Compute abstraction (work pools). Prefect 3 abstracts the compute plane behind work pools: one pool per compute type (process, Docker, Kubernetes, ECS, Cloud Run, Vertex, AWS Batch, serverless), workers poll the pool for scheduled flow runs, and a deployment names the pool it targets. Airflow's KubernetesExecutor / CeleryExecutor are a coarser abstraction — one executor per Airflow deployment, chosen at install time. Interviewers probe this axis because "how would you route dev flows to a cheap pool and prod flows to k8s?" separates people who have shipped from people who have skimmed the tutorial.
• Cost of ownership. Prefect Cloud gives you the control plane (UI, scheduler, events store, secrets) for a per-workspace / per-user fee; self-hosted OSS is free but you run the Postgres, the API server, and the UI. Airflow's OSS story is heavier (MWAA / Composer for managed); Dagster+ has a similar Cloud vs OSS split. Interviewers probe TCO because "did you actually run this thing on-call?" is the shibboleth for orchestrator questions.

The 2024 Prefect 3 rewrite — what actually changed from Prefect 2.

• Agents are gone. Prefect 2 had agents (per-queue puller processes) and workers; Prefect 3 has only workers. A worker binds to a work pool, pulls scheduled flow runs, and provisions infrastructure per run. One concept, not two.
• Sync-first execution. Prefect 2's engine was async-first; every flow ran in an event loop even if you never wrote await. Prefect 3 defaults to sync — @flow def my_flow(): ... runs like a normal Python function unless you opt into async. This removed a class of "why is my task running twice?" bugs.
• Deployments are Pythonic and IaC-friendly. In Prefect 2 you scripted deployments via Deployment.build_from_flow(); in Prefect 3 you either call flow.deploy(...) in Python or ship a prefect.yaml file and run prefect deploy. YAML-first is the interview answer.
• Result persistence + caching. Prefect 3 makes result persistence (write task outputs to a filesystem block) and caching (skip a task if inputs match a cache key) first-class. Prefect 2 had both but as bolt-ons; 3 puts them in the decorator surface.
• Events + automations. Prefect 2 had "notifications"; Prefect 3 has a full event store and an automations engine with compound triggers. This is the biggest new-in-3 capability and the one interviewers ask about most.

The Prefect vs Dagster positioning — "just Python" vs software-defined assets.

• Prefect is functional. A flow is a function; tasks are functions; you compose by calling. The mental model is "Python with orchestration decorators." No new nouns beyond flow, task, deployment, work pool.
• Dagster is asset-centric. A Dagster asset is a materialised object (a table, a model, a file) with a lineage graph, IO managers, and a materialisation function. The mental model is "declare the assets; let Dagster figure out the DAG." Great for lakehouse asset lineage; overkill for imperative ETL.
• When to pick which. Prefect for imperative, dynamic, eventful workflows (streaming triggers, on-demand runs, ad-hoc backfills). Dagster for asset-first lakehouses where the whole team thinks in tables. Airflow for legacy DAGs where the ecosystem of operators is the deciding factor.

Prefect Cloud vs self-hosted — the operator's choice.

• Prefect Cloud. SaaS control plane hosted by Prefect. You install a worker (a pip install prefect && prefect worker start process) inside your VPC; the worker calls out to Cloud to pull runs. Cloud stores flow-run metadata, exposes the UI, evaluates automations, and holds secrets. Pricing scales per-user + per-flow-run tier.
• Self-hosted OSS. Deploy prefect server (an ASGI app backed by Postgres) inside your VPC. You own the Postgres, the API server, the UI, and the events store. Free in software cost; expensive in on-call load. Sensible when your compliance regime forbids control-plane egress, or when you have SREs sitting idle.
• Hybrid pattern. Most teams start on Cloud (lower time-to-first-flow), migrate to self-hosted once they hit ~1M flow runs/month or their compliance team asks. Migration is portable — the flow / deployment / work pool concepts are unchanged.

What interviewers listen for.

• Do you name all four primitives — flows, tasks, deployments, work pools — without prompting? — senior signal.
• Do you say "agents are gone; only workers pull from work pools" when Prefect 2 → 3 comes up? — required answer.
• Do you push back on "just use Airflow" with the dynamic + eventful axis — "does Airflow give me sub-second event-driven triggers on a compound condition?" — senior signal.
• Do you name the automations engine as the differentiator against Airflow's Datasets/callbacks, not as a "nice to have"? — senior signal.
• Do you describe Prefect as "a Python-first orchestrator with work-pool-abstracted compute and an events-driven reactive engine" rather than as vague "Airflow alternative"? — required answer.

Worked example — the four-axis Prefect vs Airflow comparison

Detailed explanation. The single most useful artifact for a modern orchestrator interview is a memorised 4×2 comparison of Prefect 3 vs Airflow 3 across the four axes above. Every senior orchestration discussion converges on this table within the first five minutes; having it in your head is what separates a fluent answer from a stumbling one. Walk through building the table for a hypothetical "hourly Snowflake ingestion + event-driven ML backfill" workload.

• Workload A. Hourly ETL from Postgres to Snowflake — 40 tables, 30 min p99 runtime, static schedule.
• Workload B. Event-driven ML feature backfill — triggered when an upstream dbt run succeeds, fans out over the changed models.
• Compute. Kubernetes for the ETL; a small process pool for the ML feature job (which mostly calls a warehouse).
• Ownership target. One data-platform SRE; can't afford a second.

Question. Build the four-axis comparison for this workload and pick the orchestrator.

Input.

Axis	Prefect 3	Airflow 3
Dynamic DAGs	native; loops + conditionals in flow body	dynamic task mapping (constrained)
Eventful triggers	native automations on typed events	Datasets + sensors (lower-level)
Compute abstraction	work pools per compute type	one executor per install (mostly)
Migration cost from Airflow	rewrite DAGs as flows (~1 day per DAG)	zero (already there)

Code.

# Same "hourly ETL for one table" — read the shape, not the syntax

# Prefect 3 style
from prefect import flow, task

@task(retries=3, retry_delay_seconds=[10, 30, 60])
def extract(table: str) -> list[dict]:
    ...

@task
def load_to_snowflake(table: str, rows: list[dict]) -> int:
    ...

@flow(name="hourly_ingest_orders")
def hourly_ingest(table: str = "orders") -> int:
    rows = extract(table)
    return load_to_snowflake(table, rows)

# Airflow 3 style (for contrast)
from airflow.decorators import dag, task
from pendulum import datetime

@dag(schedule="@hourly", start_date=datetime(2026, 1, 1), catchup=False)
def hourly_ingest_orders():

    @task(retries=3)
    def extract(table: str) -> list[dict]:
        ...

    @task
    def load_to_snowflake(table: str, rows: list[dict]) -> int:
        ...

    load_to_snowflake("orders", extract("orders"))

hourly_ingest_orders()

Step-by-step explanation.

1. The Prefect flow is a plain Python function decorated with @flow; the tasks are plain functions decorated with @task. You call the flow at runtime (locally or from a scheduled deployment); there is no top-level "instantiate the DAG object" side effect.
2. The Airflow DAG is a function decorated with @dag; you must instantiate the DAG at module import time so Airflow's scheduler discovers it. Import-time side effects are the shape that makes dynamic DAGs painful in Airflow — every import scans every DAG file.
3. Retries in Prefect are per-task and support a list of backoff intervals directly on the decorator. Airflow's retries use a single retry_delay + optional exponential-backoff toggle; Prefect's list form is more expressive for "back off aggressively after the third attempt."
4. Scheduling in Prefect is a deployment attribute, not a flow attribute — flow.deploy(schedule=...). Airflow bakes the schedule into the DAG. Prefect's split lets one flow have many schedules (a "dev" hourly deployment and a "prod" 5-minute deployment) without touching the flow body.
5. For workload B (event-driven ML backfill), Prefect's automation engine subscribes to the prefect.flow-run.Completed event on the dbt_run flow and triggers the backfill directly. Airflow's Datasets accomplish the same shape but require both sides to declare the dataset URI; Prefect just watches the event bus.

Output.

Workload	Winner	Why
Hourly ETL (static, wide fan-out)	Tie	Both handle this fine
Event-driven ML backfill	Prefect 3	Native automations on typed events
SRE headcount = 1	Prefect 3	Cloud tier + zero agents/executors to run
Existing 500-DAG Airflow deployment	Airflow 3	Migration cost too high; use dynamic task mapping

Rule of thumb. Never pick an orchestrator based on "which one is trendy." Pick it based on (dynamic × eventful × compute abstraction × migration cost) — the four axes. Write the comparison on a whiteboard first; the winner falls out of the constraints, and for greenfield deployments in 2026 the answer is Prefect more often than not.

Worked example — what interviewers actually probe

Detailed explanation. The senior data-engineering Prefect interview has a predictable structure: the interviewer opens with an ambiguous question ("how would you orchestrate this pipeline?"), then progressively narrows to test whether you know the four axes and the Prefect 3 primitives. The candidates who name the primitives in sentence one score highest; the candidates who describe "we'd write a DAG" score lowest. Walk through the interview grading rubric.

• Ambiguous opener. "How would you orchestrate a nightly Snowflake ingestion?" — invites you to name flows and deployments.
• Follow-up 1. "What if we also want to run it whenever a Kafka topic gets a data-ready message?" — probes the automations / events axis.
• Follow-up 2. "How would you run this on Kubernetes in prod but on a local process in dev?" — probes the work-pool abstraction.
• Follow-up 3. "What retry / cache story would you build?" — probes task-level guarantees.
• Follow-up 4. "How do you migrate a 200-DAG Airflow deployment to Prefect?" — probes migration awareness.

Question. Draft a 5-minute senior Prefect answer that covers all four axes without waiting to be asked.

Input.

Interview signal	Weak answer	Senior answer
Primitive named	"we'd write a DAG"	"an @flow with @task steps deployed to a k8s work pool"
Eventful trigger	"we'd add a cron"	"an automation trigger on the data-ready custom event"
Compute abstraction	"one Kubernetes cluster"	"one k8s work pool for prod, one process pool for dev"
Retries + caching	"put a try/except in the task"	"@task(retries=3, retry_delay_seconds=[10,30,60], cache_key_fn=hash_inputs)"
Migration	"convert DAGs one by one"	"start with net-new flows; leave legacy DAGs on Airflow; add automations that bridge both"

Code.

Senior Prefect answer template (5 minutes)
==========================================

Minute 1 — name the primitives up front
  "I'd model this as a Prefect 3 @flow calling three @task steps —
   extract, transform, load — deployed via prefect.yaml to a
   Kubernetes work pool for prod."

Minute 2 — deployments + work pools
  "The deployment is versioned in the same Git repo as the flow;
   prefect deploy pushes it. In dev we point the same flow at a
   'process' work pool (runs on the developer's laptop). In prod
   the k8s work pool provisions one pod per flow run, uses IAM roles
   for warehouse access, and scales down to zero between runs."

Minute 3 — retries, caching, and result persistence
  "Each task carries @task(retries=3, retry_delay_seconds=[10,30,60])
   for transient warehouse errors, cache_key_fn=hash_inputs for
   idempotency across restarts, and result_storage pointing at S3
   so downstream tasks can read parent-task outputs even across
   worker restarts."

Minute 4 — eventful triggers + automations
  "We wire an automation on the 'data-ready' custom event — the
   upstream Kafka producer emits via prefect.events.emit_event(),
   the automation triggers a deployment run within seconds. We also
   set a compound-condition automation: 'if the daily ingest hasn't
   completed by 09:00 UTC → notify oncall Slack.'"

Minute 5 — migration + ownership
  "For the 200-DAG Airflow deployment we'd leave it running,
   add Prefect for net-new flows, and use automations to trigger
   Airflow via the REST API when a Prefect flow completes.
   Full migration is ~1 day per DAG on average; we budget it as
   a 6-month backlog, not a big-bang cutover."

Step-by-step explanation.

1. Minute 1 is the crucial framing. Naming the primitives immediately — "an @flow with @task steps deployed to a work pool" — signals you're a decision-maker who has actually used Prefect, not a task-runner who has read the marketing page.
2. Minute 2 addresses the compute-abstraction axis before the interviewer asks. This preempts the common trap where you commit to Prefect, then admit you don't know how to run it on Kubernetes. Naming the process pool for dev and the k8s pool for prod is senior signal.
3. Minute 3 covers the reliability guarantees — retries, caching, result persistence — and shows you understand that Prefect's decorator surface is the API for those. Naming the retry-delay list ([10, 30, 60]) rather than a single number shows fluency.
4. Minute 4 is the eventful-triggers probe — the shape Airflow doesn't do well. Naming the custom-event emit + automation trigger + compound-condition SLA is the answer that scores highest on the "why not Airflow" follow-up.
5. Minute 5 covers migration — the reliability of your judgement about when to migrate. Refusing a big-bang cutover in favour of gradual coexistence is senior signal; committing to a full port in Q1 is not.

Output.

Grading criterion	Weak score	Senior score
Names primitives in minute 1	rare	mandatory
Names work-pool split dev/prod	rare	required
Names retries + caching decorator syntax	occasional	mandatory
Names automations + custom events	rare	senior signal
Refuses big-bang migration	rare	senior signal

Rule of thumb. The senior Prefect answer is a 5-minute monologue that covers all four axes without waiting for the follow-ups. Rehearse it once; deploy it every time. If the interviewer follows up with "and Dagster?" the honest answer is "Dagster if the org already thinks in software-defined assets; Prefect otherwise."

Worked example — the "pick the orchestrator" decision tree

Detailed explanation. Given a new workload, the senior architect runs a 4-question decision tree in their head. Codifying the tree makes the interview answer reproducible: any interviewer can hand you a scenario and you can walk the tree out loud. Walk through the tree with three canonical scenarios: greenfield event-driven ML pipeline, mature 500-DAG Airflow shop, asset-first lakehouse team.

• Q1. Is the workload dynamic (fan-outs over unknown-at-parse-time lists, runtime conditionals)? → yes = Prefect (or Dagster); no = go to Q2.
• Q2. Is the workload eventful (sub-second reactive triggers on custom conditions)? → yes = Prefect; no = go to Q3.
• Q3. Does the team already think in asset lineage (dbt-style materialisations)? → yes = Dagster; no = go to Q4.
• Q4. Is there an existing Airflow deployment > 100 DAGs? → yes = extend Airflow; no = Prefect (for greenfield).

Question. Walk the decision tree for the three scenarios and record the orchestrator each ends up with.

Input.

Scenario	Q1 (dynamic?)	Q2 (eventful?)	Q3 (asset-first?)	Q4 (legacy Airflow?)
Greenfield event-driven ML	yes	yes	no	no
500-DAG Airflow shop	no	maybe	no	yes
Asset-first lakehouse	no	no	yes	no

Code.

# Decision-tree helper (illustrative)
def pick_orchestrator(dynamic: bool,
                      eventful: bool,
                      asset_first: bool,
                      legacy_airflow: bool) -> str:
    """Return the primary orchestrator for a workload."""
    if dynamic or eventful:
        return "Prefect 3"
    if asset_first:
        return "Dagster"
    if legacy_airflow:
        return "Airflow 3 (extend existing)"
    return "Prefect 3"  # greenfield default in 2026


# Walk the three scenarios
print(pick_orchestrator(True,  True,  False, False))
# → 'Prefect 3'

print(pick_orchestrator(False, False, False, True))
# → 'Airflow 3 (extend existing)'

print(pick_orchestrator(False, False, True,  False))
# → 'Dagster'

Step-by-step explanation.

1. Scenario 1 — a greenfield ML pipeline that fans out over new models emitted by upstream dbt run, triggers within seconds of the emit, and runs sub-flow-per-model. Q1 = yes → Prefect. This is the modern default: dynamic + eventful + no legacy inertia.
2. Scenario 2 — a mature Airflow shop with 500 DAGs, mostly static schedules, mostly Snowflake / dbt. Q1 = no, Q2 = maybe, Q3 = no, Q4 = yes → extend Airflow. Migration is a real cost; without a strong dynamic/eventful shape driving it, staying is the correct answer.
3. Scenario 3 — a lakehouse team where every table is an asset, every asset has upstream/downstream lineage, and the platform team wants a single graph of materialisations. Q3 = yes → Dagster. Prefect can model assets too, but Dagster's SDA model is the more natural fit.
4. The tree is deliberately not "always pick Prefect." Orchestrator choice is context-dependent; senior signal is naming a scenario in which you'd pick something else.
5. If none of Q1-Q4 clearly pass, the 2026 greenfield default is Prefect — smaller mental surface, native dynamic + eventful support, cheaper to run.

Output.

Scenario	Orchestrator	Reason
Greenfield event-driven ML	Prefect 3	dynamic + eventful axes both fire
500-DAG Airflow shop	Airflow 3	migration cost dominates
Asset-first lakehouse	Dagster	SDA model fits the team's mental model

Rule of thumb. The four-question decision tree is a whiteboard-friendly answer. Practice walking it end-to-end so an interviewer can hand you any scenario and get an orchestrator name in under 60 seconds. Prefect is the greenfield default, Airflow the legacy-extension default, Dagster the asset-first default.

Senior interview question on Prefect 3 architecture

A senior interviewer often opens with: "You inherit a small startup's Airflow 2 deployment — 30 DAGs, cron-scheduled, running on a single EC2 with LocalExecutor. The team wants sub-second event-driven runs for a new fraud-detection pipeline. Walk me through why you'd introduce Prefect 3 alongside Airflow, how you'd split responsibilities, and what the migration plan for the existing 30 DAGs would look like."

Solution Using Prefect 3 alongside Airflow — event-driven flows first, gradual migration second

# 1. New fraud-detection flow (Prefect 3, event-driven)
from prefect import flow, task
from prefect.events import emit_event

@task(retries=3, retry_delay_seconds=[10, 30, 60])
def score_transaction(txn_id: str) -> dict:
    """Call the fraud-scoring model; retry on transient failures."""
    ...

@task
def persist_score(txn_id: str, score: dict) -> None:
    """Write the scored transaction into Postgres."""
    ...

@flow(name="fraud_score_txn")
def fraud_score_txn(txn_id: str) -> dict:
    score = score_transaction(txn_id)
    persist_score(txn_id, score)
    if score["risk"] > 0.9:
        emit_event(event="fraud.high-risk", resource={"txn_id": txn_id})
    return score

# 2. prefect.yaml — versioned deployment against a k8s work pool
name: fraud
prefect-version: 3.0.0

deployments:
  - name: fraud_score_txn_prod
    entrypoint: flows/fraud.py:fraud_score_txn
    work_pool:
      name: k8s-prod
      job_variables:
        image: registry.internal/fraud:{{ get_flow_run_id() }}
        cpu: "500m"
        memory: "1Gi"
    triggers:
      - type: event
        match:
          prefect.resource.id: prefect.event.txn.received
        parameters:
          txn_id: "{{ event.payload.txn_id }}"

# 3. Custom-event bridge — Kafka consumer emits into Prefect's event bus
from prefect.events.clients import PrefectEventsClient
from kafka import KafkaConsumer

def bridge_kafka_to_prefect():
    consumer = KafkaConsumer("txn.received", bootstrap_servers="kafka:9092")
    client = PrefectEventsClient()
    for msg in consumer:
        client.emit(event="txn.received",
                    resource={"prefect.resource.id": "prefect.event.txn.received"},
                    payload={"txn_id": msg.value.decode()})

# 4. Bridge to legacy Airflow — automation triggers Airflow DAG on flow completion
# (configured in Prefect UI or via prefect.yaml automations block)
# trigger: prefect.flow-run.Completed AND flow.name == 'fraud_score_txn'
# action: send-webhook POST https://airflow.internal/api/v1/dags/downstream/dagRuns

Step-by-step trace.

Step	Before (Airflow-only)	After (hybrid)
Fraud pipeline latency	poll-based; ~1 min	event-driven; ~200 ms
Retries	dag_default_args	per-task with backoff list
Compute	single EC2 LocalExecutor	k8s work pool (pod-per-run)
Downstream trigger	Airflow ExternalTaskSensor	Prefect automation → webhook
Legacy DAG owner	Airflow (unchanged)	Airflow (unchanged)
New flow owner	(none)	Prefect (new)
Migration timeline	N/A	~6 months, gradual

After the rollout, the fraud pipeline fires within 200 ms of a txn.received Kafka message, the 30 legacy DAGs keep running on Airflow untouched, and any DAG that used to poll the fraud output now receives a webhook from a Prefect automation as soon as the flow completes. The team ships the new pipeline in Prefect within a sprint; the migration of the 30 legacy DAGs happens as a background project over 6 months.

Output:

Metric	Before	After
Fraud-score latency (event → persist)	~60 s (cron poll)	~200 ms (event)
Fraud pipeline compute	shared EC2	k8s pod-per-run
Legacy DAG count on Airflow	30	30 (unchanged)
Net-new flows on Prefect	0	1 (grows to N)
Cross-orchestrator handoff	ExternalTaskSensor	Prefect automation webhook

Why this works — concept by concept:

• @flow + @task decorators — Prefect's entire flow-definition API is two decorators wrapping ordinary Python. No YAML DSL, no imported DAG object. The fraud pipeline is a plain function; it can be unit-tested by calling fraud_score_txn("test") directly.
• prefect.yaml deployment — the deployment is a versioned config in Git that binds the flow entrypoint to a work pool, image tag, and event trigger. prefect deploy reads it and registers the deployment with the API. This is the IaC-native answer.
• k8s work pool + job_variables — the pool is the compute abstraction; job_variables let the deployment override CPU / memory / image per run without changing the pool. Workers polling the pool provision one pod per flow run and shut it down after — zero idle infrastructure.
• Automation + custom event — the deployment's triggers block subscribes to a custom txn.received event; the Kafka bridge emits into Prefect's event bus. The automation engine matches the event, extracts txn_id from the payload, and starts a flow run. Airflow's Datasets can approximate this but require both sides to declare a dataset URI.
• Cost — one Prefect Cloud workspace (~$40/mo starter), one k8s work pool (existing cluster), one worker pod (< $10/mo), one Kafka bridge process. Compared to a full Airflow migration (weeks of engineer time), the hybrid approach ships a net-new capability in a sprint while leaving the legacy DAGs untouched. Net cost of the migration is amortised over months, not front-loaded.

Design
Topic — design
Design problems on orchestration architecture

Practice →

SQL Topic — sql SQL problems on data-pipeline correctness

Practice →

---

2. Flows, tasks & subflows

@flow wraps a Python function; @task wraps its subroutines — futures, retries, caching, and result persistence live on the decorator

The mental model in one line: a prefect flow is an ordinary Python function decorated with @flow whose subroutines are decorated with @task, and every task call inside the flow returns a PrefectFuture that resolves lazily — the decorators layer retries, caching, result persistence, and state hooks on top without changing the function body, so a well-written Prefect flow reads exactly like the ETL script you'd have written anyway, only with observability, restart-safety, and per-task guarantees pinned into the framework. Every senior Prefect answer starts here; the deployment / work-pool / automations story is meaningless until the @flow / @task mental model is solid.

The four axes for flows and tasks.

• Decorator surface. @flow(name=..., retries=..., retry_delay_seconds=..., description=..., version=..., timeout_seconds=..., log_prints=...) for the flow; @task(name=..., retries=..., retry_delay_seconds=..., cache_key_fn=..., cache_expiration=..., result_storage=..., persist_result=..., tags=...) for the task. Every observability / reliability knob is declarative on the decorator.
• Futures. Every @task call inside a @flow returns a PrefectFuture — a lazy handle to the result. Task-to-task dependencies are inferred from the future being passed as an argument (load(transform(extract())) schedules extract → transform → load). Call .result() to block for the value; skip .result() and Prefect resolves lazily at the next await / return.
• Nested flows (subflows). A @flow called from inside another @flow becomes a subflow — a first-class run in the API with its own retries, result, and lineage. Subflows are the modularity primitive: extract a self-contained block of tasks into its own flow, call it from the parent, and get separate observability without giving up composition.
• State hooks. Every task and flow can attach hooks: on_completion=[log_success], on_failure=[page_oncall], on_crashed=[cleanup_temp_files]. Hooks run in-process after the state transition, get the run context, and are the correct place for side effects that must happen exactly once per state.

The retry / caching / persistence trio — three decorator knobs that carry most of the reliability weight.

• Retries. @task(retries=3, retry_delay_seconds=[10, 30, 60]) — retries the task up to 3 times with a list of backoff intervals (10s, 30s, 60s). List form beats scalar for real workloads: you want gentle initial retries, aggressive later. retry_condition_fn lets you conditionalise retry (retry on HTTPError but not on ValueError).
• Caching. @task(cache_key_fn=task_input_hash, cache_expiration=timedelta(hours=6)) — computes a cache key from the task inputs (via task_input_hash or a custom function); if a prior run produced a result for that key within the expiration window, Prefect returns the cached result instead of re-running the task. Powers idempotent restarts, backfill dedupe, and per-parameter cache reuse across flow runs.
• Result persistence. @task(persist_result=True, result_storage=S3Bucket.load("prefect-results")) — writes the task's return value to a filesystem block (S3, GCS, Azure Blob, local FS). Downstream tasks and later flow runs can fetch the persisted result via the run's UUID; caching depends on it.

Subflows — the modularity primitive.

• When to use. Any coherent block of tasks that (a) needs its own retries / caching / result persistence policy, (b) needs to be reused across multiple parent flows, or (c) is worth surfacing as a separate run in the UI. Subflows show up as their own row in the flow-run list.
• Return values. A subflow's return value is a PrefectFuture (async subflow) or the value directly (sync subflow). Compose subflow returns exactly like task returns.
• Nested subflows. Subflows can call other subflows; there is no depth limit. The lineage graph in the UI is a tree.
• Anti-pattern. Using subflows just to "group" tasks visually. That's what tags are for. Subflows imply separate retry / caching / result policy — if you don't need that, keep the tasks in the parent.

Result persistence — where task outputs live, and why you care.

• The default. Task results live in worker memory until the flow finishes; then they're gone. Fine for small, fast flows; catastrophic for long-running ones where a worker crash forces a full replay.
• Persisted results. persist_result=True writes the return value (pickled by default) to the configured result store. On restart, downstream tasks fetch persisted upstream results and skip re-running finished tasks.
• Result storage blocks. S3Bucket, GCS, Azure, LocalFileSystem. Blocks are named credential + config objects registered in the API; the decorator references them by name so credentials never live in code.
• Serialisation. Default is PickleSerializer. For interoperability, use JSONSerializer (typed dicts only) or a custom serialiser. Pandas DataFrames need pandas-aware serialisation to survive Parquet round-trips.

Common interview probes on flows and tasks.

• "What's the difference between @flow and @task?" — required answer: flow is the top-level entry (or subflow); task is a step inside it with retries / caching / observability.
• "How do you retry a task on a specific exception?" — retry_condition_fn on the decorator, checking isinstance(exception, X).
• "How does Prefect handle worker crashes mid-task?" — with persist_result=True, upstream results survive; the crashed task retries from scratch; downstream is skipped until the retry succeeds.
• "When would you use a subflow vs a task?" — subflow when you want separate retries / caching / result policy or reuse across parents; task otherwise.

Worked example — a full ETL flow with tasks, retries, caching, and result persistence

Detailed explanation. The canonical Prefect 3 ETL flow: an @flow calls three @task subroutines (extract → transform → load) with retries, exponential backoff, cache keys pinned to the input parameters, and result persistence to S3. On worker crash, upstream results survive and only the failing task retries. Walk through every knob.

• Extract. Reads Postgres, returns list of dicts. Retries on transient DB errors.
• Transform. Pandas dataframe ops, cached on (source_hash, transform_version).
• Load. Writes to Snowflake, retries on network errors, no cache (side-effect only).
• Result persistence. All task outputs written to s3://prefect-results/etl/<flow_run_id>/.

Question. Write the flow, tasks, and the block configuration for S3 result storage.

Input.

Component	Value
Flow	ingest_orders(source_hash: str, target_table: str)
Tasks	extract, transform, load
Retry policy	retries=3, retry_delay_seconds=[10, 30, 60]
Cache	transform only; cache_key_fn=task_input_hash
Result storage	S3Bucket.load("prefect-results")

Code.

from datetime import timedelta
from prefect import flow, task
from prefect.tasks import task_input_hash
from prefect_aws import S3Bucket

# One-time block registration (run in setup script or Prefect UI):
# S3Bucket(bucket_name="prefect-results",
#          aws_access_key_id=..., aws_secret_access_key=...).save("prefect-results")

RESULT_STORE = S3Bucket.load("prefect-results")

@task(
    retries=3,
    retry_delay_seconds=[10, 30, 60],
    persist_result=True,
    result_storage=RESULT_STORE,
    tags=["extract"],
)
def extract(source_hash: str) -> list[dict]:
    """Read the orders delta from Postgres for the given source_hash slice."""
    import psycopg2
    conn = psycopg2.connect("host=db-primary port=5432 dbname=production user=cdc_reader")
    with conn.cursor() as cur:
        cur.execute("""
            SELECT id, customer_id, total_cents, status, updated_at
            FROM   public.orders
            WHERE  source_hash = %s
        """, (source_hash,))
        rows = [{"id": r[0], "customer_id": r[1], "total_cents": r[2],
                 "status": r[3], "updated_at": r[4]} for r in cur.fetchall()]
    conn.close()
    return rows

@task(
    retries=2,
    retry_delay_seconds=[5, 15],
    cache_key_fn=task_input_hash,
    cache_expiration=timedelta(hours=6),
    persist_result=True,
    result_storage=RESULT_STORE,
    tags=["transform"],
)
def transform(rows: list[dict], transform_version: str = "v3") -> list[dict]:
    """Normalise + enrich rows; cached by (input_hash, transform_version)."""
    import pandas as pd
    df = pd.DataFrame(rows)
    df["total_dollars"] = df["total_cents"] / 100
    df["status"] = df["status"].str.upper()
    df["transform_version"] = transform_version
    return df.to_dict(orient="records")

@task(
    retries=5,
    retry_delay_seconds=[10, 30, 60, 120, 300],
    tags=["load"],
)
def load(rows: list[dict], target_table: str) -> int:
    """Load rows to Snowflake; no caching — side effect only."""
    import snowflake.connector
    conn = snowflake.connector.connect(
        user=..., password=..., account=..., warehouse="ETL_WH", database="ANALYTICS"
    )
    with conn.cursor() as cur:
        cur.executemany(
            f"MERGE INTO {target_table} AS tgt USING VALUES ... ",
            rows,
        )
        n = cur.rowcount
    conn.close()
    return n

@flow(name="ingest_orders", log_prints=True)
def ingest_orders(source_hash: str, target_table: str = "ANALYTICS.ORDERS") -> int:
    """Extract → transform → load, with full observability + restart safety."""
    rows        = extract(source_hash)
    normalised  = transform(rows)
    loaded      = load(normalised, target_table)
    print(f"Loaded {loaded} rows into {target_table}")
    return loaded

Step-by-step explanation.

1. The three tasks are ordinary Python functions decorated with @task. Every reliability knob — retries, delay list, caching, result persistence, tags — is declared on the decorator, keeping the function body free of framework code. This is the "just Python" positioning in action.
2. retry_delay_seconds=[10, 30, 60] is the list form: attempt 1 waits 10s, attempt 2 waits 30s, attempt 3 waits 60s. Scalar retry_delay_seconds=30 gives a fixed 30s between retries; the list is preferred for transient errors that may need longer settling time.
3. The transform task's cache_key_fn=task_input_hash builds a cache key from the task's input arguments (rows + transform_version). If a previous run cached this exact input within 6 hours, Prefect returns the cached result instead of re-running the transform. Perfect for backfills where the same input is re-processed multiple times.
4. persist_result=True writes the return value to the S3 result store keyed by the run's UUID. If a worker crashes after transform succeeds but before load starts, the retry can fetch the persisted transform result and skip re-running it — an enormous savings on expensive tasks.
5. The @flow body is imperative Python. Each @task call returns a PrefectFuture; passing the future as the next task's argument creates the dependency edge. print() output is captured to the flow logs because log_prints=True is set on the decorator.

Output.

Stage	Result	Cached?	Persisted?
extract("2026-07-21-slice-01")	12,345 dicts	no	yes
transform(rows, "v3")	12,345 enriched dicts	yes (6h)	yes
load(rows, "ANALYTICS.ORDERS")	12,345 rows merged	no	no (side effect only)
Total flow duration	~90 s
Worker-crash recovery	resumes from load

Rule of thumb. For every production task, decide three things up front: (a) is it idempotent (safe to retry) — set retries + retry_delay_seconds list; (b) is it deterministic on inputs — set cache_key_fn=task_input_hash; (c) is its output needed downstream after a crash — set persist_result=True + a result storage block. Skipping any of these ships a fragile flow.

Worked example — subflows for a modular ingestion pipeline

Detailed explanation. A parent flow orchestrates ingestion for six source systems; each source has its own extract / transform / load logic. Rather than putting all six sets of tasks in one flow, factor each source's pipeline into its own subflow. The parent flow calls the six subflows (optionally in parallel via .submit()); each subflow appears as a separate run in the UI with independent retries, caching, and lineage. Walk through the pattern.

• Parent flow. ingest_all_sources — loops over source configs, calls one subflow per source.
• Subflow. ingest_one_source — the reusable extract / transform / load block.
• Failure isolation. One source's failure does not stop the others; parent flow reports partial success.
• Observability. Each source's run is a first-class row in the UI.

Question. Refactor a monolithic 6-source ingestion into a parent flow that calls six subflows.

Input.

Component	Value
Parent flow	ingest_all_sources(source_configs: list[dict])
Subflow	ingest_one_source(source: dict)
Sources	postgres_orders, mysql_customers, salesforce_leads, ...
Parallelism	up to 4 concurrent subflows
Failure policy	continue on subflow failure; report per-source status

Code.

from prefect import flow, task
from prefect.states import State
from prefect.futures import PrefectFuture

@task(retries=3, retry_delay_seconds=[10, 30, 60])
def extract_one(source: dict) -> list[dict]:
    """Source-specific extract logic; dispatch by source['kind']."""
    if source["kind"] == "postgres":
        return _extract_postgres(source)
    elif source["kind"] == "mysql":
        return _extract_mysql(source)
    elif source["kind"] == "salesforce":
        return _extract_salesforce(source)
    raise ValueError(f"unknown source kind: {source['kind']}")

@task
def transform_one(rows: list[dict], source: dict) -> list[dict]:
    ...

@task(retries=5, retry_delay_seconds=[10, 30, 60, 120, 300])
def load_one(rows: list[dict], target: str) -> int:
    ...

@flow(name="ingest_one_source", retries=1)
def ingest_one_source(source: dict) -> int:
    """One-source ETL — extract, transform, load. Retryable at flow level."""
    rows       = extract_one(source)
    normalised = transform_one(rows, source)
    loaded     = load_one(normalised, source["target_table"])
    return loaded

@flow(name="ingest_all_sources", log_prints=True)
def ingest_all_sources(source_configs: list[dict]) -> dict[str, int | str]:
    """Parent flow — fan out to N subflows; continue on subflow failure."""
    results: dict[str, int | str] = {}
    for src in source_configs:
        try:
            loaded = ingest_one_source(src)
            results[src["name"]] = loaded
        except Exception as e:
            print(f"[WARN] source {src['name']} failed: {e}")
            results[src["name"]] = f"failed: {e}"
    return results

Step-by-step explanation.

1. Each source's pipeline lives inside ingest_one_source, decorated as a @flow. When called from ingest_all_sources, it becomes a subflow — a first-class flow run in the Prefect API with its own retries, caching, and lineage. The subflow's retries=1 policy is separate from the parent's.
2. The parent flow loops over source_configs and calls the subflow synchronously; each subflow's completion (or failure) is recorded as a distinct row in the UI. This is the "observability without gluing tasks together" benefit that pushes teams to subflows.
3. The try / except around the subflow call ensures one source's failure doesn't abort the others. The parent flow returns a per-source dict — a source succeeded with a row count or a source failed with an error message. Consumers can decide whether to alert on any failure or only on threshold.
4. For actual parallelism (not sequential subflow calls), wrap each subflow call in .submit() (in an async parent flow) or use a ConcurrentTaskRunner. The pattern above is sequential; parallel is a small extension.
5. Subflows are the reuse primitive too: ingest_one_source can be triggered directly (from a deployment, an automation, or the UI) with just one source, without touching the parent flow. This is what makes subflows the correct modularity choice — reusability plus independent policy.

Output.

Source	Rows loaded	Status
postgres_orders	12,345	success
mysql_customers	4,567	success
salesforce_leads	0	failed: HTTPError 502
stripe_charges	23,412	success
shopify_orders	8,910	success
segment_events	145,678	success

Rule of thumb. Use subflows whenever a coherent block of tasks (a) has its own retry / caching policy, (b) is worth surfacing as an independent run in the UI, or (c) needs to be triggered directly. Do not use subflows purely for visual grouping — tags accomplish that at lower cost. The subflow overhead is one extra flow-run row per call; that's a real cost for tight inner loops.

Senior interview question on Prefect flows and tasks

A senior interviewer might ask: "Design a Prefect 3 flow that ingests 50 tables from Postgres to Snowflake nightly, with per-table retries, cache-driven idempotent backfill, and worker-crash recovery that avoids re-extracting from the source when the load fails. Include the decorator config, the result persistence choice, and the subflow-or-tasks decision for the 50 tables."

Solution Using per-table subflows with cached extracts and persisted intermediates

from datetime import timedelta
from prefect import flow, task
from prefect.tasks import task_input_hash
from prefect_aws import S3Bucket

RESULT_STORE = S3Bucket.load("prefect-results")

@task(
    retries=3,
    retry_delay_seconds=[10, 30, 60],
    cache_key_fn=task_input_hash,
    cache_expiration=timedelta(hours=24),   # per-day cache
    persist_result=True,
    result_storage=RESULT_STORE,
    tags=["extract"],
)
def extract_table(table: str, run_date: str) -> list[dict]:
    """Cached per (table, run_date); safe to re-run within 24h without re-hitting source."""
    ...

@task(
    retries=2,
    retry_delay_seconds=[5, 15],
    persist_result=True,
    result_storage=RESULT_STORE,
    tags=["transform"],
)
def normalise(rows: list[dict], table: str) -> list[dict]:
    ...

@task(
    retries=5,
    retry_delay_seconds=[10, 30, 60, 120, 300],
    tags=["load"],
)
def load_snowflake(rows: list[dict], target: str) -> int:
    ...

@flow(name="ingest_one_table", retries=1)
def ingest_one_table(table: str, run_date: str) -> int:
    """One table's ingestion; cached extract survives worker crashes for 24h."""
    rows = extract_table(table, run_date)
    norm = normalise(rows, table)
    return load_snowflake(norm, f"ANALYTICS.{table.upper()}")

@flow(name="ingest_50_tables", log_prints=True)
def ingest_50_tables(tables: list[str], run_date: str) -> dict[str, int | str]:
    """Parent flow — one subflow per table; failure isolated per-table."""
    results: dict[str, int | str] = {}
    for t in tables:
        try:
            results[t] = ingest_one_table(t, run_date)
        except Exception as e:
            results[t] = f"failed: {e}"
    return results

Step-by-step trace.

Step	Config	Behaviour
extract_table cache	cache_expiration=timedelta(hours=24)	Same (table, run_date) hit within 24h → cached
extract_table retries	retries=3 + [10,30,60]	Transient Postgres errors absorbed
normalise persistence	persist_result=True	Survives worker crash before load
load_snowflake retries	retries=5 + backoff to 5 min	Snowflake network flakes absorbed
Subflow retries	retries=1 on ingest_one_table	Whole-table retry if any task fails
Parent failure isolation	try/except in ingest_50_tables	One bad table doesn't abort the run

After the deployment, a worker crash mid-run resumes with all completed extracts cached (no re-hitting Postgres) and all persisted normalisations available (no re-computing). The nightly run for 50 tables completes in ~40 minutes with typical wall-clock, ~15 minutes on a crash-recovery re-run because most cached extracts hit.

Output:

Metric	Value
Tables ingested per run	50
Extract cache hit rate on retry	> 90%
Worker-crash recovery savings	~60% wall-clock
Per-table failure isolation	yes
Idempotent backfill	yes (cache)

Why this works — concept by concept:

• Per-table subflows — each table's ETL is a separate flow run in the UI, with its own retries. The parent flow's try/except isolates failures so one bad table doesn't abort the other 49. Observability + isolation for the price of one decorator.
• cache_key_fn=task_input_hash + 24h expiration — turns the extract_table task into an idempotent, backfill-friendly primitive. Re-runs within 24h hit cache; the source Postgres sees each (table, run_date) pair at most once.
• persist_result=True + S3 block — worker crashes are recoverable without re-computing upstream tasks. The intermediate normalise output is pickled to S3; on retry, downstream tasks fetch the persisted value.
• Retry-delay list — [10, 30, 60] for extract, [5, 15] for transform, [10, 30, 60, 120, 300] for load. The pattern matches the failure mode: shorter waits for cheap ops, longer waits for network-bound ops with harder-to-recover-from transient failures.
• Cost — one S3 bucket for result storage (~pennies/mo for typical Prefect run volumes), one Prefect Cloud workspace, cached extracts drop 90%+ of source-DB load on retries. Compared to a naive "just retry the whole flow" design, the per-task cache + persistence eliminates most re-work. O(distinct-inputs) per day rather than O(retries × distinct-inputs).

SQL
Topic — sql
SQL problems on incremental ingestion and idempotency

Practice →

Design Topic — design Design problems on pipeline retry and caching

Practice →

---

3. Deployments + work pools

A deployment is a versioned flow config; a work pool is the compute abstraction a worker binds to — decouple the flow from where it runs

The mental model in one line: a prefect deployment is a durable, versioned binding of (a flow entrypoint + parameters + schedule + triggers + work pool + job variables) stored in Prefect's API, and a prefect work pool is a named compute abstraction (process, docker, kubernetes, ecs, cloud-run, vertex, serverless) that workers poll for scheduled runs — the two together decouple "what runs" from "where it runs," so one flow can have a dev deployment on a local process pool, a staging deployment on Docker, and a prod deployment on Kubernetes, all from one prefect.yaml file. This decoupling is the single biggest ergonomic win Prefect 3 has over Airflow, and it's the shape senior interviewers probe hardest.

The four axes for deployments and work pools.

• Versioning. Deployments are versioned in Prefect's API — each prefect deploy creates or updates a deployment with a version string (default: git SHA or timestamp). Rollback = pin an older version. Interviewers probe versioning because "how do I roll back a bad ETL change at 3 AM?" is the shape that separates junior from senior orchestration answers.
• IaC ergonomics. prefect.yaml at the repo root declares all deployments; prefect deploy applies them. The YAML lives in Git next to the flow code — the deployment is source-controlled, code-reviewed, and CI-deployable exactly like the flow. Airflow's DAG-as-Python-file has similar-ish semantics but the deployment metadata is baked into the DAG.
• Work pool as compute abstraction. A work pool is a type (process, docker, kubernetes, ecs-task, cloud-run-v2, vertex-ai, push-to-serverless) plus base job variables (image, resources, IAM role, network). Workers bind to the pool and provision infrastructure per run. Multiple deployments can share one pool; multiple pools can share one flow.
• Worker vs agent (the Prefect 2 → 3 change). Prefect 2 had agents — long-running processes that pulled from queues. Prefect 3 replaces them with workers, which pull from work pools. The mental model is simpler: one primitive, not two. prefect worker start --pool <name> is the entire runtime command.

The prefect.yaml — the deployment IaC file.

• What lives there. One deployments: entry per deployment: name, entrypoint (file.py:flow_function), parameters, work_pool + job_variables (image, cpu, memory, env), schedules (cron or interval), triggers (event-based).
• What lives elsewhere. Secrets (in Prefect blocks or your secrets manager), block configuration (via prefect block ls), work-pool creation (prefect work-pool create <name> --type <type>).
• Multi-environment. One repo can carry dev / staging / prod deployments in one prefect.yaml with different work pools per entry; CI applies the right subset per environment.
• Templating. {{ }} Jinja substitutions for environment variables, git metadata, and run-time context. Powers "image tag = current git SHA" without a shell script.

Work-pool types — the compute abstraction menu.

• process. Worker executes the flow in a child process on the same host. Cheap, no isolation. For dev, local testing, and small internal jobs.
• docker. Worker starts a Docker container per flow run using a specified image. Isolation, dependency pinning, no cluster. For single-host or docker-swarm deployments.
• kubernetes. Worker submits a Job per flow run into a namespace. Full isolation, cluster autoscaling, RBAC. The default choice for production.
• ecs-task / cloud-run-v2 / vertex-ai / push-to-serverless. Managed serverless pools; the worker submits a task/service definition to AWS ECS, GCP Cloud Run, GCP Vertex, or a serverless provider. No always-on worker cost for push pools; the platform provisions the compute on demand.

Schedules and triggers — first-class deployment attributes.

• Cron schedules. schedule: cron: "0 * * * *" — the classic. Every deployment can carry zero, one, or many schedules.
• Interval schedules. schedule: interval: 300 — every 5 minutes, no cron parsing.
• RRule schedules. schedule: rrule: "..." — iCal-format for complex "third Tuesday of the month" patterns.
• Event triggers. triggers: [{ type: event, match: {...}, parameters: {...} }] — subscribe to a Prefect event; when it matches, start a run with the templated parameters. The eventful-orchestration primitive.

Common interview probes on deployments and work pools.

• "What's the difference between a flow and a deployment?" — required answer: flow is the code; deployment is the versioned binding of (flow, work pool, schedule, parameters, triggers) that the API knows about.
• "Why did Prefect kill agents in Prefect 3?" — one primitive, not two; workers pull from work pools; simpler operational surface.
• "How do you run the same flow on Kubernetes in prod and locally in dev?" — two deployments with different work_pool.name values in prefect.yaml.
• "How do you roll back a deployment?" — pin an older version; or re-run prefect deploy from the older git SHA.

Worked example — a full prefect.yaml for dev/staging/prod deployments

Detailed explanation. The canonical multi-environment prefect.yaml: one flow, three deployments (dev / staging / prod), three work pools (process / docker / kubernetes), Jinja templating for image tags, and event triggers on the prod deployment. Walk through every block.

• Flow. flows/ingest.py:ingest_orders — one Python entrypoint.
• Dev deployment. Runs on a local process pool; no schedule; developer triggers ad-hoc.
• Staging deployment. Runs on a Docker pool; hourly schedule; image = git SHA.
• Prod deployment. Runs on a k8s pool; hourly schedule; event trigger on custom data-ready event.

Question. Write the prefect.yaml for the three deployments, plus the shell commands to create the work pools and start the workers.

Input.

Environment	Work pool	Schedule	Trigger
dev	dev-process (process)	none	manual
staging	staging-docker (docker)	hourly	none
prod	k8s-prod (kubernetes)	hourly	data-ready event

Code.

# prefect.yaml — repo root
name: order-ingestion
prefect-version: 3.0.0

# --- Build / push image once; shared by staging + prod --------------------
build:
  - prefect_docker.deployments.steps.build_docker_image:
      id: build-image
      image_name: registry.internal/ingest
      tag: "{{ get_current_commit_sha() }}"
      dockerfile: Dockerfile

push:
  - prefect_docker.deployments.steps.push_docker_image:
      image_name: "{{ build-image.image_name }}"
      tag: "{{ build-image.tag }}"

# --- Deployments ----------------------------------------------------------
deployments:

  - name: ingest_orders_dev
    entrypoint: flows/ingest.py:ingest_orders
    parameters:
      target_table: "DEV.ORDERS"
    work_pool:
      name: dev-process

  - name: ingest_orders_staging
    entrypoint: flows/ingest.py:ingest_orders
    parameters:
      target_table: "STAGING.ORDERS"
    work_pool:
      name: staging-docker
      job_variables:
        image: "{{ build-image.image_name }}:{{ build-image.tag }}"
    schedules:
      - cron: "0 * * * *"        # hourly

  - name: ingest_orders_prod
    entrypoint: flows/ingest.py:ingest_orders
    parameters:
      target_table: "ANALYTICS.ORDERS"
    work_pool:
      name: k8s-prod
      job_variables:
        image: "{{ build-image.image_name }}:{{ build-image.tag }}"
        cpu: "1000m"
        memory: "2Gi"
        env:
          SNOWFLAKE_ROLE: ETL_ROLE
    schedules:
      - cron: "0 * * * *"
    triggers:
      - type: event
        match:
          prefect.resource.id: "prefect.event.data-ready"
        parameters:
          target_table: "ANALYTICS.ORDERS"

# One-time — create the three work pools
prefect work-pool create dev-process    --type process
prefect work-pool create staging-docker --type docker
prefect work-pool create k8s-prod       --type kubernetes

# One-time — apply the yaml
prefect deploy --all

# Start workers (one per pool; scale k8s worker via Deployment replicas)
prefect worker start --pool dev-process        # runs on developer laptop
prefect worker start --pool staging-docker     # runs on the staging box
prefect worker start --pool k8s-prod           # runs inside the k8s cluster as a Deployment

Step-by-step explanation.

1. The build: step at the top of prefect.yaml builds a Docker image once per prefect deploy invocation. The tag {{ get_current_commit_sha() }} pins the image to the current git commit, giving reproducible rollback: git checkout <old-sha> && prefect deploy re-deploys the exact old version.
2. Each deployments: entry binds the same flow entrypoint (flows/ingest.py:ingest_orders) to a different work pool + schedule + parameters. The flow code doesn't know it's being deployed three ways — that's the point of the abstraction.
3. job_variables: under work_pool: override the pool's default job template per-deployment. Prod gets more CPU/memory and a specific IAM role via the env block; dev gets nothing (uses process pool defaults). One flow, three infrastructure profiles.
4. The schedules: block accepts a list; a single deployment can carry multiple schedules (hourly and nightly, for example). Cron is the most common; interval and rrule are alternatives for simpler or more complex cadences.
5. The triggers: block on the prod deployment subscribes to a custom data-ready event; when the event matches, Prefect starts a flow run in addition to the cron schedule. This is the eventful-orchestration primitive — cron + event triggers coexist on one deployment.

Output.

Deployment	Compute	Schedule	Trigger
ingest_orders_dev	local process	none (manual)	none
ingest_orders_staging	docker container	0 * * * *	none
ingest_orders_prod	k8s job (1 CPU / 2Gi)	0 * * * *	event data-ready

Rule of thumb. For every net-new flow, ship three deployments (dev / staging / prod) in one prefect.yaml from day one — even if only prod is scheduled. The two extra entries cost nothing to define, and they force you to name work pools per environment, which is the shape that catches "we forgot to isolate dev traffic" bugs.

Worked example — Kubernetes work pool with pod-per-run and cluster autoscaling

Detailed explanation. The canonical production Prefect setup: a Kubernetes work pool that spawns one pod per flow run, tied to a cluster autoscaler that scales node count up during a run and down between runs. Zero idle worker cost; pay only for compute during flow execution. Walk through the pool config, the base job template, and the autoscaler assumptions.

• Pool. k8s-prod, type kubernetes, namespace prefect-runs.
• Base job template. Pod spec with resource requests, service account, image pull secrets, tolerations for a prefect-runs node pool.
• Worker. One long-running Deployment inside the cluster; polls the pool, submits Jobs.
• Autoscaler. Cluster autoscaler scales the prefect-runs node pool from 0 to 20 based on pending pods.

Question. Configure the k8s work pool, the base job template, and the worker Deployment.

Input.

Component	Value
Namespace	prefect-runs
Node pool	prefect-runs (tainted; only Prefect pods)
Base CPU / mem	500m / 1Gi (deployment-overridable)
Service account	prefect-flow-sa with warehouse IAM binding
Autoscaler min / max	0 / 20 nodes

Code.

# Base job template — set via `prefect work-pool inspect k8s-prod`
apiVersion: batch/v1
kind: Job
metadata:
  generateName: "prefect-{{ flow_run.deployment_id }}-"
  namespace: prefect-runs
  labels:
    prefect.io/flow-run-id: "{{ flow_run.id }}"
    prefect.io/deployment-name: "{{ flow_run.deployment_name }}"
spec:
  ttlSecondsAfterFinished: 3600
  backoffLimit: 0                        # Prefect handles retries; k8s doesn't
  template:
    spec:
      restartPolicy: Never
      serviceAccountName: prefect-flow-sa
      tolerations:
        - key: dedicated
          operator: Equal
          value: prefect-runs
          effect: NoSchedule
      nodeSelector:
        node-pool: prefect-runs
      containers:
        - name: flow
          image: "{{ image }}"           # from job_variables
          resources:
            requests:
              cpu: "{{ cpu | default('500m') }}"
              memory: "{{ memory | default('1Gi') }}"
            limits:
              cpu: "{{ cpu | default('2000m') }}"
              memory: "{{ memory | default('4Gi') }}"
          env:
            - name: PREFECT_API_URL
              value: "{{ PREFECT_API_URL }}"
            - name: PREFECT_API_KEY
              valueFrom:
                secretKeyRef:
                  name: prefect-api-key
                  key: token

# Worker Deployment — one long-running pod polls the work pool
apiVersion: apps/v1
kind: Deployment
metadata:
  name: prefect-worker-k8s-prod
  namespace: prefect-runs
spec:
  replicas: 1                            # scale higher if flow-run rate demands
  selector:
    matchLabels:
      app: prefect-worker
  template:
    metadata:
      labels:
        app: prefect-worker
    spec:
      serviceAccountName: prefect-worker-sa
      containers:
        - name: worker
          image: prefecthq/prefect:3-latest
          args: ["prefect", "worker", "start", "--pool", "k8s-prod", "--type", "kubernetes"]
          env:
            - name: PREFECT_API_URL
              value: "https://api.prefect.cloud/api/accounts/.../workspaces/..."
            - name: PREFECT_API_KEY
              valueFrom:
                secretKeyRef:
                  name: prefect-api-key
                  key: token
          resources:
            requests:
              cpu: "100m"
              memory: "256Mi"

Step-by-step explanation.

1. The base job template is the pod spec Prefect submits for each flow run. generateName ensures every Job gets a unique name; ttlSecondsAfterFinished: 3600 garbage-collects the Job (and its logs metadata) an hour after completion so the cluster doesn't drown in finished Jobs.
2. backoffLimit: 0 disables Kubernetes-level retries — Prefect handles retries at the task and flow level, and duplicating that at the k8s level double-counts failures. If the pod crashes, Prefect marks the flow run as failed and applies its own retry policy.
3. The tolerations + nodeSelector pin flow-run pods to a dedicated prefect-runs node pool. This isolates data-plane workloads from the rest of the cluster (no noisy neighbours) and lets the cluster autoscaler manage that pool independently, scaling from 0 to 20 nodes based on pending Prefect pods.
4. The worker Deployment is the puller: one long-running pod (100m CPU idle) that polls the k8s-prod work pool for scheduled flow runs and submits Kubernetes Jobs. The worker itself does not run flow code; it's a control-plane process.
5. job_variables: in the deployment's prefect.yaml (e.g. image, cpu, memory) substitute into the base template at run time — so one pool can serve deployments with wildly different resource profiles without needing multiple pools.

Output.

Aspect	Value
Idle worker cost	~$5/mo (one 100m/256Mi pod)
Idle flow-run cost	$0 (pods only exist during runs)
Cluster autoscale range	0-20 nodes
Max concurrent flow runs	~40 (2 pods per node × 20 nodes)
Per-flow-run overhead	~5 s (pod scheduling)
Retry semantics	Prefect-side only (backoffLimit=0)

Rule of thumb. For any Prefect Kubernetes work pool, set backoffLimit: 0 (Prefect owns retries), ttlSecondsAfterFinished to 1-6 hours (kill zombie Jobs), pin to a dedicated node pool via tolerations + nodeSelector (isolate from other workloads), and scale the worker replicas by expected concurrency (one worker pod handles ~50 concurrent flow runs comfortably).

Worked example — schedules, triggers, and the "one flow, many deployments" pattern

Detailed explanation. A single flow (report_daily_kpis) needs three run modes: (1) 08:00 UTC every day for the exec dashboard, (2) 12:00 UTC every day for the EMEA team, (3) on-demand whenever a dbt-completed event fires. The Prefect answer is one flow, three deployments, one work pool. Walk through the pattern and the YAML.

• Flow. report_daily_kpis(region: str = "GLOBAL") — parameterised by region.
• Deployment A. Global report at 08:00 UTC (cron).
• Deployment B. EMEA report at 12:00 UTC (cron with different parameter).
• Deployment C. On-demand triggered by dbt-completed event.

Question. Write the three deployments in one prefect.yaml and explain the parameter-injection story.

Input.

Deployment	Schedule / Trigger	Parameters
report_daily_kpis_global	0 8 * * *	region="GLOBAL"
report_daily_kpis_emea	0 12 * * *	region="EMEA"
report_daily_kpis_ondemand	event dbt-completed	region="{{ event.payload.region }}"

Code.

# prefect.yaml
name: reporting
prefect-version: 3.0.0

deployments:
  - name: report_daily_kpis_global
    entrypoint: flows/reports.py:report_daily_kpis
    parameters:
      region: "GLOBAL"
    work_pool:
      name: k8s-prod
    schedules:
      - cron: "0 8 * * *"
        timezone: "UTC"

  - name: report_daily_kpis_emea
    entrypoint: flows/reports.py:report_daily_kpis
    parameters:
      region: "EMEA"
    work_pool:
      name: k8s-prod
    schedules:
      - cron: "0 12 * * *"
        timezone: "UTC"

  - name: report_daily_kpis_ondemand
    entrypoint: flows/reports.py:report_daily_kpis
    work_pool:
      name: k8s-prod
    triggers:
      - type: event
        match:
          prefect.resource.id: "prefect.event.dbt-completed"
        parameters:
          region: "{{ event.payload.region }}"

Step-by-step explanation.

1. All three deployments point at the same flow entrypoint (flows/reports.py:report_daily_kpis). The flow doesn't know it's being deployed three ways — the deployment layer is entirely orthogonal to the flow definition. This is the "decouple what runs from when/where it runs" primitive.
2. Each deployment injects different parameters: — region="GLOBAL" for the 08:00 run, region="EMEA" for the 12:00 run. Prefect merges deployment-level parameters with the flow's default arguments; per-run overrides (from the UI or API) win over both.
3. Deployment C has no schedules: block but has a triggers: block. When Prefect's event bus receives a dbt-completed event, the trigger fires and starts a flow run with region extracted from the event payload via the {{ event.payload.region }} template.
4. All three deployments share the same k8s-prod work pool. Multiple deployments per pool is the norm; sharing a pool is the cheap default. Splitting pools is only worth it when compute profiles or scheduling policies genuinely differ.
5. This pattern is impossible in Airflow without three separate DAG files (or a lot of dynamic-DAG gymnastics). Prefect's flow / deployment split is the answer: flow definition is stable; deployments proliferate as business needs demand.

Output.

Deployment	Fires when	Runs as
report_daily_kpis_global	08:00 UTC daily	report_daily_kpis(region="GLOBAL")
report_daily_kpis_emea	12:00 UTC daily	report_daily_kpis(region="EMEA")
report_daily_kpis_ondemand	dbt-completed event	report_daily_kpis(region=<from event>)

Rule of thumb. One flow, many deployments — never the other way around. When you need a new schedule, parameter, or trigger, add a deployment, not a flow. This keeps the flow definition minimal and the deployment YAML the record of "what business intent lives on top of this flow."

Senior interview question on deployments and work pools

A senior interviewer might ask: "Design a Prefect 3 deployment strategy for a team of 20 data engineers shipping ~100 flows across three environments (dev / staging / prod) into a Kubernetes cluster. Cover the work pool layout, the prefect.yaml per-repo pattern, the image build / push story, the secrets flow, and the rollback playbook when a bad deploy ships at 3 AM."

Solution Using per-repo prefect.yaml, shared work pools, git-SHA-tagged images, and version-pinned rollback

# 1. Global work-pool layout — created once by the platform team
# prefect work-pool create dev-process     --type process
# prefect work-pool create staging-docker  --type docker
# prefect work-pool create k8s-prod        --type kubernetes
# prefect work-pool create k8s-dev         --type kubernetes  # cheap, single-node

# 2. Per-repo prefect.yaml — one file per code repo, checked in to Git
name: fraud-detection
prefect-version: 3.0.0

build:
  - prefect_docker.deployments.steps.build_docker_image:
      id: build-image
      image_name: registry.internal/fraud-detection
      tag: "{{ get_current_commit_sha() }}"

push:
  - prefect_docker.deployments.steps.push_docker_image:
      image_name: "{{ build-image.image_name }}"
      tag: "{{ build-image.tag }}"

deployments:
  - name: fraud_score_dev
    entrypoint: flows/fraud.py:fraud_score_txn
    work_pool: { name: k8s-dev, job_variables: { image: "{{ build-image.image_name }}:{{ build-image.tag }}", cpu: "200m", memory: "512Mi" } }

  - name: fraud_score_staging
    entrypoint: flows/fraud.py:fraud_score_txn
    work_pool: { name: k8s-prod, job_variables: { image: "{{ build-image.image_name }}:{{ build-image.tag }}", cpu: "500m", memory: "1Gi", env: { STAGE: staging } } }
    schedules: [{ cron: "*/15 * * * *" }]

  - name: fraud_score_prod
    entrypoint: flows/fraud.py:fraud_score_txn
    work_pool: { name: k8s-prod, job_variables: { image: "{{ build-image.image_name }}:{{ build-image.tag }}", cpu: "1000m", memory: "2Gi", env: { STAGE: prod } } }
    schedules: [{ cron: "*/5 * * * *" }]
    triggers: [{ type: event, match: { prefect.resource.id: "prefect.event.txn-received" }, parameters: { txn_id: "{{ event.payload.txn_id }}" } }]

# 3. CI job — apply deployments on every merge to main
prefect cloud login --key $PREFECT_API_KEY --workspace $PREFECT_WORKSPACE
prefect deploy --all --prefect-file prefect.yaml

# 4. Rollback playbook — pin an older git SHA in seconds
# a) Find the last-known-good SHA in git log
BAD_SHA=$(git rev-parse HEAD)
GOOD_SHA=$(git log --format=%H --skip=1 -1)

# b) Check out the good SHA and re-deploy
git checkout $GOOD_SHA
prefect deploy --all --prefect-file prefect.yaml

# c) Verify next scheduled run picks up the older image tag
prefect deployment inspect "fraud_score_prod"

Step-by-step trace.

Layer	Config	Purpose
Work pools	4 pools (dev-process, staging-docker, k8s-dev, k8s-prod)	shared across all 100 flows
Per-repo prefect.yaml	one per code repo	keeps deploys close to code
Image tag	{{ get_current_commit_sha() }}	reproducible rollback via git
CI deploy	prefect deploy --all on merge	GitOps for deployments
Rollback	git checkout + re-deploy	~30 s recovery
Secrets	Prefect blocks + k8s secrets	never in prefect.yaml
Observability	one workspace, tagged by repo	one UI, per-repo filter

After adoption, the platform team owns the four work pools and the k8s-prod worker Deployments; each product team owns its own prefect.yaml. Deploys are GitOps — merge to main triggers a CI job that runs prefect deploy; rollback is git checkout <old-sha> && prefect deploy, taking under 30 seconds. Image tags are git SHAs, so any deployment can be traced back to exact source code.

Output:

Metric	Value
Work pools	4 (platform-managed)
Deployments	~300 (100 flows × 3 envs)
Deploy cadence	on every merge to main
Rollback time	< 30 s (git + prefect deploy)
Idle worker cost	~$5/mo per pool
Per-flow-run isolation	full (k8s Job per run)

Why this works — concept by concept:

• Per-repo prefect.yaml + shared work pools — product teams own their deployments and flows in their own repos; the platform team owns the compute abstraction (work pools). Clear ownership boundary; no cross-team coupling on infrastructure.
• Git-SHA image tag — every deployment carries an exact git commit as its image tag. Rollback is git checkout <old-sha> && prefect deploy — under 30 seconds, no image rebuild needed (the old image is already in the registry).
• GitOps deploy on merge — CI runs prefect deploy --all on every merge to main. Deployments live in Git; the API is the eventual state; drift between the two is a bug. Same operating model as Kubernetes manifests + ArgoCD, but for orchestration.
• Prefect blocks for secrets — API keys, warehouse credentials, S3 access keys live in Prefect blocks (or in k8s secrets referenced from the base job template). Never in prefect.yaml. Rotation is a one-place change.
• Cost — 4 work pools (one worker per pool, ~$5/mo each = $20/mo), no per-flow-run idle cost (k8s Jobs terminate after runs), no per-deployment fee. Compared to Airflow's per-DAG-file scanning and static-pool sizing, Prefect's pool + deployment model is dramatically cheaper at 100+ flows. O(1) infrastructure per flow.

Design
Topic — design
Design problems on multi-environment deployment

Practice →

Streaming Topic — streaming Streaming problems on event-driven pipelines

Practice →

---

4. Automations, events & triggers

automations turn Prefect into a reactive orchestration engine — typed events, compound-condition triggers, and Slack / email / PagerDuty actions

The mental model in one line: every state transition in Prefect 3 (a flow run starting, completing, crashing; a deployment being created; a worker going offline) emits a typed event onto Prefect's event bus, and an automation is a durable rule that matches events against a compound-condition trigger (match + within + threshold) and fires an action — sending a Slack message, paging on-call, running another deployment, cancelling a run, or emitting a downstream event — which turns Prefect from a scheduler into a reactive orchestration engine and answers the "how do I know when my pipeline broke?" question with a first-class API instead of on_failure callbacks and TODO comments. This is the biggest capability new in Prefect 3 and the shape that most changes what an orchestrator can do.

The four axes for automations and events.

• Events. Structured JSON records: event name (e.g. prefect.flow-run.Completed), resource (the object it applies to, e.g. a specific flow run), payload (arbitrary custom data), occurred timestamp. Prefect emits system events automatically; your code can emit custom events via emit_event().
• Triggers. The condition side of an automation. Types: event (fire on a matching event), metric (fire when a metric crosses a threshold), compound (multiple sub-triggers with AND/OR), sequence (event X then event Y in order). Every trigger supports within (time window) and threshold (count).
• Actions. The response side. Types: send-notification (Slack, email, PagerDuty via a Prefect block), run-deployment (start another flow), pause-deployment / resume-deployment, cancel-flow-run, change-flow-run-state, send-event (emit an event to trigger another automation), send-webhook (call an external URL).
• Latency. Prefect Cloud evaluates automations in the control plane with a target latency of ~seconds from event emission to action fired. Self-hosted latency depends on your Postgres and the worker/scheduler cadence.

System events — the automatic emissions.

• Flow-run lifecycle. prefect.flow-run.Pending, Running, Completed, Failed, Crashed, Cancelled, TimedOut. One per state transition.
• Task-run lifecycle. Same shape, per task run. High volume — filter carefully.
• Deployment lifecycle. Created, Updated, Deleted.
• Worker lifecycle. WorkerOnline, WorkerOffline — useful for "worker down for > 5 min → page on-call."
• Block / block-document lifecycle. For observability into credential rotations, block updates, etc.

Custom events — the extension point.

• What. Any event your code emits via emit_event(event="data-ready", resource={...}, payload={...}). Not restricted to Prefect's built-in event names.
• Where. From inside a flow (from prefect.events import emit_event) or from an external process (via the PrefectEventsClient). External emit is how bridges (Kafka consumer, webhook receiver, cron job) inject events into Prefect's bus.
• Why. Custom events are how you turn a domain signal ("dbt run completed," "S3 file arrived," "Kafka topic reached watermark") into a Prefect trigger without polling.

Compound triggers — the "and / or" primitives.

• AND. type: compound, require: all, triggers: [ trigger_A, trigger_B ] — fire only when both A and B match within a time window.
• OR. type: compound, require: any, triggers: [ trigger_A, trigger_B ] — fire on either.
• Sequence. type: sequence, triggers: [ trigger_A, trigger_B ] — fire when A happens then B happens, in order.
• Threshold + within. threshold: 3, within: 5m — fire when the matching event count reaches 3 within 5 minutes. The pattern for "3 failures in 5 min → circuit breaker."

SLA automations — the highest-value shape.

• The pattern. "If deployment X hasn't emitted a Completed event by 09:00 UTC, notify #data-oncall." Encodes an SLA as a first-class automation instead of a downstream health-check job.
• The primitives. A sequence or compound trigger with an absence check (via posture: proactive in some Cloud tiers) or a scheduled probe-flow that emits a sla-miss event which then triggers a notification.
• Why it beats sensors. Airflow sensors sit inside a DAG and consume a worker slot; Prefect SLA automations sit in the control plane and consume nothing until they fire.

Common interview probes on automations and events.

• "What's a Prefect event?" — required answer: a typed JSON record emitted on every state transition or custom emit_event() call.
• "How do you build an SLA alert?" — automation with a trigger on the absence of flow-run.Completed by a target time; action send-notification to a Slack block.
• "How does the automation engine differ from Airflow callbacks?" — Prefect automations are control-plane rules, durable across worker restarts, and support compound conditions; Airflow callbacks run in-DAG and can miss on scheduler crashes.
• "How do you avoid alert-flood on flapping failures?" — threshold: N within: 5m on the trigger.

Worked example — a Slack notification on flow-run failure

Detailed explanation. The most common automation: send a Slack message whenever any flow run in the workspace fails. Requires (a) a SlackWebhook block with the incoming-webhook URL, (b) an automation with a flow-run.Failed trigger, and (c) a send-notification action referencing the block. Walk through the setup.

• Block. SlackWebhook.load("data-oncall-webhook") — one-time registration.
• Trigger. flow-run.Failed on any flow.
• Action. send-notification with a templated body naming the flow and the run URL.

Question. Configure the block, the automation, and the notification body.

Input.

Component	Value
Slack channel	#data-oncall
Block name	data-oncall-webhook
Event trigger	prefect.flow-run.Failed
Filter	all flows in workspace
Notification body	flow name, run URL, run ID

Code.

# 1. One-time block registration (Python or Prefect UI)
from prefect_slack import SlackWebhook

SlackWebhook(url="https://hooks.slack.com/services/...").save("data-oncall-webhook")

# 2. Automation config (Prefect Cloud UI or `prefect automation create` YAML)
name: notify_on_flow_failure
description: Slack #data-oncall on any flow-run failure
trigger:
  type: event
  posture: reactive
  match:
    prefect.resource.id: "prefect.flow-run.*"
  expect:
    - "prefect.flow-run.Failed"
  within: 0                              # fire immediately
  threshold: 1
actions:
  - type: send-notification
    block_document_id: <slack-webhook-block-uuid>
    subject: "Prefect flow failed: {{ flow.name }}"
    body: |
      Flow: {{ flow.name }}
      Deployment: {{ deployment.name | default('(ad-hoc)') }}
      Run: {{ flow_run.name }}
      Run URL: {{ flow_run.ui_url }}
      Message: {{ flow_run.state.message | default('(none)') }}

# 3. Alternative — create the automation via Python
from prefect.automations import Automation
from prefect.events.schemas.automations import EventTrigger, Posture
from prefect.blocks.notifications import SlackWebhook

slack = SlackWebhook.load("data-oncall-webhook")

Automation(
    name="notify_on_flow_failure",
    trigger=EventTrigger(
        posture=Posture.Reactive,
        expect={"prefect.flow-run.Failed"},
        match={"prefect.resource.id": "prefect.flow-run.*"},
        within=0,
        threshold=1,
    ),
    actions=[
        slack.notify(
            subject="Prefect flow failed: {{ flow.name }}",
            body="Run URL: {{ flow_run.ui_url }}"
        )
    ]
).create()

Step-by-step explanation.

1. The SlackWebhook block stores the incoming-webhook URL as a Prefect secret. Blocks decouple credentials from code — the automation references the block by name, so the URL never lives in prefect.yaml or in git.
2. The EventTrigger with expect: [prefect.flow-run.Failed] matches any event of that name on any flow-run resource. posture: reactive means "fire when the event arrives"; the alternative proactive posture fires on the absence of the event within a window (used for SLAs).
3. within: 0 + threshold: 1 means "fire immediately on the first matching event." For flap protection on a chatty flow, bump threshold: 5 + within: 5m — fire only when 5 failures happen in 5 minutes.
4. The send-notification action templates the message with {{ flow.name }}, {{ deployment.name }}, {{ flow_run.ui_url }} — Prefect substitutes these from the event's resource context. The Jinja templating is the same language used in prefect.yaml.
5. On Prefect Cloud, the automation is evaluated in the control plane within ~seconds of the event emission. On self-hosted, the API server evaluates automations on the same polling cadence as the scheduler.

Output.

Event	Automation fires?	Slack message posted?
flow-run.Failed on any flow	yes	yes
flow-run.Completed	no	no
flow-run.Crashed	no (different event)	no
flow-run.Failed × 3 in 5 min	yes × 3 (no threshold)	3 messages
Add threshold: 3, within: 5m	fires once after 3rd failure	1 message

Rule of thumb. Every production Prefect workspace needs three baseline automations from day one: (1) flow-run failure → Slack, (2) worker offline for > 5 min → PagerDuty, (3) any deployment paused → Slack. All three are one YAML block each; skipping any of them leaves your ops team blind.

Worked example — SLA automation with the "expected duration" pattern

Detailed explanation. The SLA shape: "the daily order-ingest flow must complete within 30 minutes of its scheduled start; if not, page the on-call engineer." Encoded as a compound automation: (a) trigger on flow-run.Running for the deployment, (b) start a 30-minute timer, (c) if flow-run.Completed doesn't arrive within the window, fire a send-notification action to PagerDuty. Walk through the pattern.

• Deployment. ingest_orders_prod on the k8s-prod pool.
• SLA window. 30 minutes from start.
• Escalation. PagerDuty page on breach.
• Runbook link. Included in the payload for the on-call.

Question. Build the SLA automation with the compound trigger and the PagerDuty action.

Input.

Component	Value
Deployment	ingest_orders_prod
SLA window	30 min
Escalation	PagerDuty
Runbook URL	https://runbooks.internal/ingest_orders
Threshold	1 breach → page

Code.

# Automation: SLA breach on ingest_orders_prod
name: sla_ingest_orders_prod
description: Page oncall if the daily ingest doesn't complete within 30 min
trigger:
  type: sequence
  within: 1800s                                    # 30 min window
  triggers:
    - type: event
      posture: reactive
      match:
        prefect.resource.id: "prefect.deployment.<deployment-uuid>"
      expect:
        - "prefect.flow-run.Running"
    # If the corresponding Completed event does NOT arrive in the window,
    # the sequence is considered breached and the actions fire.
    - type: event
      posture: proactive
      match:
        prefect.resource.id: "prefect.deployment.<deployment-uuid>"
      expect:
        - "prefect.flow-run.Completed"

actions:
  - type: send-notification
    block_document_id: <pagerduty-block-uuid>
    subject: "[SLA BREACH] ingest_orders_prod > 30 min"
    body: |
      Deployment: ingest_orders_prod
      Started: {{ flow_run.start_time }}
      Now: {{ occurred }}
      Duration exceeded: 30 min
      Runbook: https://runbooks.internal/ingest_orders
      Run URL: {{ flow_run.ui_url }}

# Alternative — emit a custom sla-breach event that a downstream automation reacts to
# (useful when you want the SLA logic to also cancel the run, or trigger a backup flow)
from prefect.events import emit_event

def emit_sla_breach(flow_run_id: str, deployment_name: str):
    emit_event(
        event="sla.breach",
        resource={"prefect.resource.id": f"prefect.event.sla-breach.{deployment_name}"},
        payload={"flow_run_id": flow_run_id, "deployment": deployment_name},
    )

# A second automation subscribes to `sla.breach` and:
# - cancels the offending flow run
# - triggers a backup deployment on a fallback work pool
# - pages a different on-call rotation

Step-by-step explanation.

1. The sequence trigger says "fire when event A (flow-run.Running) is not followed by event B (flow-run.Completed) within the within window (30 min)." The posture: proactive on the second trigger inverts the match — it fires on absence, not on presence.
2. The match block scopes both triggers to the specific deployment UUID — the automation doesn't fire on other deployments' runs. Every automation should be scoped as narrowly as the workflow demands to keep the event bus efficient.
3. The send-notification action posts to PagerDuty via a PagerDutyBlock block. The templated body includes the flow-run's URL, the runbook, and the wall-clock duration — everything the on-call needs to triage without switching tools.
4. The custom-event escalation pattern (the second Python snippet) is the extensible shape: emit a sla.breach event and let a second automation react. That second automation can chain actions — cancel the offending run, start a backup flow, log to a downstream analytics system. Composition via events is more flexible than a single automation with many actions.
5. On Prefect Cloud, SLA-shaped automations (posture: proactive) are evaluated by the control plane's scheduler on a ~1 s cadence; latency from breach to action fired is typically 1-5 seconds.

Output.

Scenario	Automation state	Action fires?
Ingest completes in 15 min	sequence satisfied	no
Ingest completes in 29:59	sequence satisfied (just)	no
Ingest still running at 30:00	sequence breached	PagerDuty page
Ingest crashes at 25:00	Completed never arrives → breach	PagerDuty page
Ingest cancelled at 20:00	Completed never arrives → breach	PagerDuty page (may not be desired)

Rule of thumb. For every production deployment, encode the SLA as a proactive-posture sequence automation. Scope the match block to the specific deployment; set the within window to 1.5× the p95 run duration; escalate to the appropriate rotation (Slack for non-critical, PagerDuty for critical). Reserving SLA logic in the orchestrator, not in downstream health-check jobs, keeps ownership clear.

Worked example — custom event emitter bridging Kafka to Prefect

Detailed explanation. A Kafka topic (orders.received) carries new orders; every new order should trigger the fraud_score_txn flow within seconds. Kafka is outside Prefect's world; the bridge is a small Python process that consumes from Kafka and calls emit_event() on Prefect's event bus. An automation on the deployment subscribes to the custom event and starts the flow. Walk through the whole pattern.

• Kafka topic. orders.received, one message per order (JSON with txn_id).
• Bridge process. Long-running Python consumer; emits orders.received custom events.
• Automation trigger. Match on the custom event; extract txn_id from payload; run fraud_score_txn deployment.
• Latency budget. ~1 s end-to-end from Kafka to flow-run start.

Question. Write the bridge process, the emit call, and the automation configuration.

Input.

Component	Value
Kafka topic	orders.received
Custom event name	orders.received
Automation deployment	fraud_score_txn_prod
Latency target	< 2 s Kafka → run start
Message payload	JSON {"txn_id": "..."}

Code.

# bridge_kafka_to_prefect.py — runs as a long-lived process (k8s Deployment)
import json
from kafka import KafkaConsumer
from prefect.events.clients import PrefectEventsClient

def main():
    consumer = KafkaConsumer(
        "orders.received",
        bootstrap_servers=["kafka:9092"],
        group_id="prefect-bridge",
        enable_auto_commit=True,
        value_deserializer=lambda v: json.loads(v.decode()),
    )
    client = PrefectEventsClient()

    for msg in consumer:
        client.emit(
            event="orders.received",
            resource={"prefect.resource.id": "prefect.event.orders-received"},
            payload={"txn_id": msg.value["txn_id"]},
        )

if __name__ == "__main__":
    main()

# Automation config
name: run_fraud_score_on_orders_received
description: Fire fraud_score_txn_prod on every orders.received custom event
trigger:
  type: event
  posture: reactive
  match:
    prefect.resource.id: "prefect.event.orders-received"
  expect:
    - "orders.received"
  within: 0
  threshold: 1
actions:
  - type: run-deployment
    deployment_id: <fraud_score_txn_prod-uuid>
    parameters:
      txn_id: "{{ event.payload.txn_id }}"

# Bridge process as a k8s Deployment (one replica; scale up for higher Kafka lag budgets)
apiVersion: apps/v1
kind: Deployment
metadata:
  name: prefect-kafka-bridge
  namespace: prefect-runs
spec:
  replicas: 1
  template:
    spec:
      containers:
        - name: bridge
          image: registry.internal/prefect-kafka-bridge:latest
          env:
            - name: PREFECT_API_URL
              value: "https://api.prefect.cloud/api/accounts/.../workspaces/..."
            - name: PREFECT_API_KEY
              valueFrom: { secretKeyRef: { name: prefect-api-key, key: token } }

Step-by-step explanation.

1. The bridge is a long-running Python process (one replica in k8s, more if Kafka lag becomes a concern). It consumes from a specific Kafka topic with a stable group_id so restarts pick up where they left off — Kafka's consumer-group offsets are the durability primitive.
2. Each Kafka message is transformed into a Prefect custom event via PrefectEventsClient.emit(). The event name (orders.received) is arbitrary — it just has to match the automation's expect clause. The resource.id scopes the event under a stable identifier the automation subscribes to.
3. The automation subscribes to resource.id: "prefect.event.orders-received" and fires on the orders.received event. within: 0, threshold: 1 = fire immediately on every message. For rate-limiting, bump the threshold.
4. The run-deployment action starts a run of the fraud_score_txn_prod deployment with parameters templated from the event payload. {{ event.payload.txn_id }} extracts the transaction ID from the emitted event and injects it as the flow's argument.
5. End-to-end latency: Kafka commit → bridge consume (~50 ms) → emit event (~50 ms over network to Prefect API) → automation match (~1 s cadence on Cloud) → run-deployment action (~500 ms) → k8s pod schedule (~5 s) → flow-run starts. Total ~7-8 s cold, ~1-2 s if the k8s worker pool has warm capacity.

Output.

Kafka message	Bridge action	Prefect event	Automation action	Flow starts within
{"txn_id":"A"}	emit orders.received	1 event	run fraud_score_txn_prod(txn_id=A)	~7 s cold, ~1 s warm
{"txn_id":"B"}	emit orders.received	1 event	run fraud_score_txn_prod(txn_id=B)	~7 s cold, ~1 s warm
Bridge restart	resumes at last offset	no gap	no gap	(no runs missed)
Kafka down	consumer retries	no events	automation quiet	0 (recovers on Kafka up)

Rule of thumb. Any external signal — Kafka message, S3 event notification, webhook, cron job — becomes a first-class Prefect trigger via a small bridge + emit_event() + an automation with run-deployment. This is the eventful-orchestration primitive that lets Prefect replace a whole layer of custom trigger glue.

Senior interview question on Prefect automations

A senior interviewer might ask: "You have three critical daily flows that must complete before 09:00 UTC for the exec dashboard. Design the automation strategy — the events, the compound triggers, the notification channels, the flap suppression, and the SLA-breach escalation. Explain how this differs from an Airflow-based approach."

Solution Using per-flow SLA automations, flap-suppressed failure notifications, and a compound "any-of-three-late" trigger

# 1. Per-flow SLA automation (repeat 3x with different deployment IDs)
name: sla_flow_A_09_utc
description: If flow A hasn't completed by 09:00 UTC, notify #data-oncall
trigger:
  type: event
  posture: proactive
  match:
    prefect.resource.id: "prefect.deployment.<flow-A-uuid>"
  expect:
    - "prefect.flow-run.Completed"
  within: 32400s                        # from 00:00 UTC (or scheduled start) to 09:00
  threshold: 1                          # need at least 1 completion in the window
actions:
  - type: send-notification
    block_document_id: <slack-oncall>
    subject: "[SLA] Flow A did not complete by 09:00 UTC"
    body: |
      Deployment: {{ deployment.name }}
      Expected: 09:00 UTC
      Actual latest run: {{ flow_run.state.name | default('(none)') }}
      Runbook: https://runbooks.internal/{{ deployment.name }}

# 2. Compound "any-of-three-late" trigger — page if ANY of the three critical
#    deployments emit an sla.breach custom event
name: page_oncall_on_any_critical_sla_breach
trigger:
  type: compound
  require: any
  within: 60s
  triggers:
    - type: event
      match: { prefect.resource.id: "prefect.event.sla-breach.flow_A" }
      expect: ["sla.breach"]
    - type: event
      match: { prefect.resource.id: "prefect.event.sla-breach.flow_B" }
      expect: ["sla.breach"]
    - type: event
      match: { prefect.resource.id: "prefect.event.sla-breach.flow_C" }
      expect: ["sla.breach"]
actions:
  - type: send-notification
    block_document_id: <pagerduty-critical>
    subject: "[CRITICAL] Exec dashboard SLA breached"
    body: |
      One or more of flow A / B / C failed to complete by 09:00 UTC.
      Runbook: https://runbooks.internal/exec-dashboard-sla

# 3. Flap-suppressed failure notification — no more than 1 page per 15 min
name: notify_on_flow_failure_flap_suppressed
trigger:
  type: event
  posture: reactive
  match:
    prefect.resource.id: "prefect.flow-run.*"
  expect: ["prefect.flow-run.Failed"]
  within: 900s                          # 15 min window
  threshold: 3                          # 3 failures → notify once
actions:
  - type: send-notification
    block_document_id: <slack-oncall>
    subject: "3+ flow failures in 15 min"
    body: "Investigate; likely a shared dependency degraded."

# 4. In each of the 3 flows: emit the sla.breach event on failure to trigger the compound automation
from prefect.events import emit_event

@flow(name="flow_A")
def flow_A():
    try:
        # ... work ...
        pass
    except Exception:
        emit_event(event="sla.breach",
                   resource={"prefect.resource.id": "prefect.event.sla-breach.flow_A"})
        raise

Step-by-step trace.

Layer	Component	Purpose
Per-flow SLA	3 proactive automations	wall-clock deadline enforcement
Compound page	1 compound automation	single page for any of three breaches
Flap suppression	reactive + threshold=3 within=15m	reduce alert fatigue
Custom event emission	in-flow emit_event()	orchestrator-independent signal
Notification channels	Slack for warnings, PagerDuty for pages	severity-tiered
Escalation	runbook URL in every notification	shortest path to fix

After deployment, each of the three critical flows carries its own SLA automation with a 09:00 UTC deadline; if any of them fail to complete, they emit an sla.breach.<flow> custom event, which the compound automation catches to page on-call once (not three times). Ordinary failures are absorbed by the flap-suppressed notification — three failures in 15 minutes ping Slack once. The on-call is not woken for a single flap.

Output:

Scenario	Automations that fire	Notifications sent
Flow A completes 08:59 UTC	none	0
Flow A fails once, retries succeed	0 (no threshold hit)	0
Flow A late past 09:00 UTC	SLA + compound	1 Slack + 1 page
All three flows late past 09:00	3× SLA + 1 compound	3 Slack + 1 page (compound has debounce)
5 unrelated flow failures in 15 min	flap-suppressed	1 Slack

Why this works — concept by concept:

• Proactive-posture SLA — the expect: [Completed] within: N fires on the absence of the event within the window. This is the correct primitive for deadline-based SLAs; a reactive-posture automation could only fire on state transitions that actually happen.
• Compound OR trigger + custom sla.breach event — every flow emits a domain-specific sla.breach.<flow> event on failure; the compound automation catches any of them with require: any. One page per exec-dashboard SLA breach, not three, even if all three flows fail.
• Flap suppression via threshold + within — threshold: 3, within: 15m means the reactive failure notification fires once after 3 failures in 15 minutes, not 3 times. This is the shape that keeps on-call responsive rather than desensitised.
• Runbook URL in every notification — the shortest path to fix is a runbook URL in the notification body. Every automation carries Runbook: https://runbooks.internal/...; on-call clicks once and lands on the mitigation steps.
• Cost — automations are free on Prefect Cloud (up to per-workspace-tier limits); notifications cost the underlying block's cost (a Slack webhook is free, PagerDuty per-page). Compared to a whole "monitoring stack" (Prometheus alerts + Alertmanager + custom bridges), Prefect automations serve the orchestration-native use case with zero extra infrastructure. O(1) alert per breach, not O(N) per underlying signal.

Streaming
Topic — streaming
Streaming problems on event-driven triggers

Practice →

Design Topic — design Design problems on SLA and alerting systems

Practice →

---

5. Prefect vs Airflow / Dagster + interview signals

Prefect wins dynamic + eventful; Airflow wins legacy ecosystem; Dagster wins asset-lineage — pick per workload

The mental model in one line: the 2026 Python orchestrator market is a three-horse race between prefect (dynamic + eventful workflows, "just Python" positioning, work-pool compute abstraction), Airflow 3 (legacy DAG scheduling, the largest operator ecosystem, TaskFlow API for modern DAGs, Datasets for asset-ish lineage), and Dagster (software-defined assets, first-class lineage graph, IO managers, embedded dbt integration) — no single orchestrator wins on every axis, so the senior answer is workload-driven: dynamic + eventful → Prefect; legacy DAGs on a mature Airflow deployment → Airflow; asset-first lakehouse team → Dagster; the migration cost from an existing orchestrator is usually the tie-breaker.

The four axes for orchestrator comparison.

• Dynamic DAG support. Prefect is native — flows are Python functions, loops and conditionals just work. Airflow supports dynamic task mapping (@task.expand(...)) but with constraints (must return before mapping, limits on cardinality). Dagster's dynamic outputs support similar shapes to Airflow's mapping.
• Eventful triggers. Prefect's automations engine is the cleanest — compound triggers on typed events. Airflow's Datasets are the closest analogue (schedule on dataset updates) but sit lower in the stack. Dagster's sensors are a middle ground — Python code that polls or subscribes, then triggers a run.
• Asset lineage. Dagster is native — assets are first-class objects with lineage, IO managers, and materialisation history. Airflow's Datasets give a coarse asset-ish lineage. Prefect can emit asset lineage but doesn't model it natively — you add it via emit_event or an external catalog.
• Ecosystem / operators. Airflow wins — the operator ecosystem is unmatched (hundreds of providers, thousands of operators). Prefect has integrations (prefect_aws, prefect_dbt, prefect_snowflake) but far fewer. Dagster has focused integrations (dbt, Spark, Snowflake) at deeper quality.

Migration paths — Prefect 1/2 → Prefect 3 breaking changes.

• Agents → workers. Prefect 2 agents are removed; replace with workers + work pools. prefect worker start --pool <name> is the replacement command.
• Deployment API. Deployment.build_from_flow() from Prefect 2 is deprecated in Prefect 3; use flow.deploy(...) in Python or prefect.yaml + prefect deploy.
• Async → sync default. Prefect 2 flows implicitly ran in an event loop; Prefect 3 defaults to sync execution unless the flow is declared async def. Existing async flows work but require explicit async annotation.
• Result persistence API. Prefect 2's storage blocks are re-shaped in Prefect 3; migrate S3Bucket config; verify result URIs.
• Automations replace notifications. Prefect 2's notifications block is superseded by the automations engine. Migrate notification rules to automations.

Prefect vs Airflow — the head-to-head.

• Prefect wins when. Workloads are dynamic (runtime-known fan-out), eventful (sub-second reactive), or need per-flow compute isolation (one k8s Job per run without an executor rebuild). Small teams that value ergonomics over ecosystem.
• Airflow wins when. You already have Airflow at scale (hundreds of DAGs, mature CI/CD, invested SREs), the operator ecosystem covers your integrations natively, or the org standard is Airflow and migration is not on the table.
• Both are fine when. The workload is medium-static ETL; either will run it. The choice is then about developer ergonomics (Prefect's edge) vs ecosystem depth (Airflow's edge).

Prefect vs Dagster — the head-to-head.

• Prefect wins when. The workload is imperative (write functions; wire them up), eventful (respond to signals), or dynamic (fan out over runtime-known lists). Teams that don't want to buy into the SDA mental model.
• Dagster wins when. The org already thinks in software-defined assets — every table is an asset, every asset has upstream / downstream lineage, dbt is central. Dagster's IO managers and asset materialisation history are unmatched here.
• Both are fine when. Mixed asset + non-asset workflows. Prefect's flow-of-tasks model absorbs asset workloads via manual lineage emission; Dagster's asset model absorbs imperative jobs via ops.

Interview probes on Prefect vs the alternatives.

• "Why not Airflow?" — required answer: dynamic + eventful shapes + per-flow compute isolation.
• "Why not Dagster?" — required answer: SDA overhead not worth it when the workload isn't asset-first.
• "What's the migration cost?" — required answer: ~1 day per DAG on average; ~1 sprint per aggregate for outbox / event patterns; gradual coexistence beats big-bang.
• "How do you observe Prefect in production?" — Prefect Cloud UI + automations + structured logs + pg_stat_statements-equivalent metrics from the Prefect API.

Worked example — the 5-axis, 3-orchestrator matrix

Detailed explanation. The definitive comparison table for the "which orchestrator" interview question. Score each axis with a dot-rating (1 = weak, 5 = strong) so you can defend the score under follow-up. Walk through each cell.

• Prefect 3 strengths. Dynamic DAGs (5), eventful triggers (5), work-pool compute abstraction (5), migration cost from Airflow (2 — real work).
• Airflow 3 strengths. Operator ecosystem (5), TaskFlow API (4), Datasets (3), migration cost to Airflow-from-nothing (5 — zero).
• Dagster strengths. Asset lineage (5), IO managers (5), dbt integration (5), migration cost from Airflow/Prefect (2 — real work).

Question. Build the 5-axis × 3-orchestrator matrix and defend the scoring.

Input.

Axis	Prefect 3	Airflow 3	Dagster
Dynamic DAGs	5	3	4
Eventful triggers	5	3	4
Asset lineage	2	2	5
Operator ecosystem	3	5	3
Migration cost from Airflow	2	5 (stay)	2

Code.

When asked "which orchestrator for X?", walk the matrix:

For X = "streaming-triggered ML backfill":
  dynamic + eventful dominate → Prefect (5+5) beats Airflow (3+3) beats Dagster (4+4)

For X = "500 static DAGs already on Airflow":
  migration cost dominates → Airflow (5 stay) beats Prefect (2) beats Dagster (2)

For X = "asset-lineage lakehouse":
  asset lineage dominates → Dagster (5) beats both (2)

For X = "greenfield hybrid batch + event":
  no single axis dominates → Prefect (25 total) beats Airflow (19) beats Dagster (18)

Step-by-step explanation.

1. Dynamic DAGs — Prefect scores 5 because flows are ordinary Python; loops, conditionals, runtime fan-outs all work naturally. Dagster scores 4 because dynamic outputs are supported but require explicit @dynamic_partitions_def decoration. Airflow scores 3 because dynamic task mapping is supported but with cardinality limits and must-return-before-scheduling constraint.
2. Eventful triggers — Prefect scores 5 because automations + typed events + compound triggers are first-class. Dagster scores 4 (sensors work well but require Python polling code). Airflow scores 3 (Datasets are the closest but sit at the DAG-schedule level; ExternalTaskSensor is the fallback and consumes a worker slot).
3. Asset lineage — Dagster scores 5 (native SDA model with IO managers, asset materialisation history, lineage graph). Prefect and Airflow both score 2 (can emit asset events but don't model lineage natively; require external catalog).
4. Operator ecosystem — Airflow scores 5 (hundreds of providers, thousands of operators, decades of community). Prefect scores 3 (~30 official integrations, growing). Dagster scores 3 (focused integrations at high quality but narrower coverage).
5. Migration cost from Airflow — Airflow scores 5 (stay put; zero migration). Prefect and Dagster both score 2 (real migration cost; ~1 day per DAG on average). This axis matters when the workload doesn't force a change.

Output.

Total scores	Prefect	Airflow	Dagster
Sum	17	19	18
Wins on	dynamic + eventful	ecosystem + migration	asset lineage
Loses on	ecosystem	dynamic + eventful	ecosystem
Sweet spot	greenfield modern	legacy at scale	asset-first lakehouse

Rule of thumb. The three orchestrators have different sweet spots; there is no universal winner. Score against your specific workload; the highest-total orchestrator on the axes you weight is your answer. Never argue "X is better" without naming the axes.

Worked example — migrating a 20-DAG Airflow deployment to Prefect

Detailed explanation. Given a mid-sized Airflow deployment (20 DAGs, mostly hourly, mostly Snowflake), plan a 6-month gradual migration to Prefect 3. Walk through the phases: bridge, greenfield-only, migrate-newest-first, deprecate-oldest-last.

• Month 1-2. Stand up Prefect Cloud workspace; add k8s work pool; convert one DAG as a pilot.
• Month 2-4. Ship all new flows on Prefect; freeze net-new development on Airflow.
• Month 4-6. Migrate remaining Airflow DAGs one by one; retire Airflow when the last DAG lands.
• Rollback plan. Keep Airflow running throughout; migration is one-way but no big-bang.

Question. Draft the 6-month migration plan with phase gates and rollback criteria.

Input.

Phase	Duration	Milestone
1. Pilot	4 weeks	1 DAG on Prefect end-to-end
2. Bridge	4 weeks	Airflow ↔ Prefect webhook bridge
3. Freeze	8 weeks	no new Airflow DAGs; new flows on Prefect only
4. Migrate	8 weeks	port 20 DAGs one by one
5. Retire	2 weeks	shut down Airflow

Code.

# Phase 2 — bridge: Airflow DAG that calls Prefect via API
from airflow.decorators import dag, task
from pendulum import datetime
import requests, os

PREFECT_API = os.environ["PREFECT_API_URL"]
PREFECT_KEY = os.environ["PREFECT_API_KEY"]

@dag(schedule=None, start_date=datetime(2026, 1, 1), catchup=False)
def bridge_to_prefect():

    @task
    def trigger_prefect_deployment(deployment_id: str, parameters: dict):
        r = requests.post(
            f"{PREFECT_API}/deployments/{deployment_id}/create_flow_run",
            json={"parameters": parameters},
            headers={"Authorization": f"Bearer {PREFECT_KEY}"},
        )
        r.raise_for_status()
        return r.json()["id"]

    trigger_prefect_deployment("<uuid>", {"key": "value"})

bridge_to_prefect()

# Reverse bridge — Prefect automation that calls Airflow's REST API on flow completion
# Configured via `send-webhook` action in the automation:
#   POST https://airflow.internal/api/v1/dags/downstream/dagRuns
#   body: {"conf": {"trigger_source": "prefect", "flow_run_id": "{{ flow_run.id }}"}}

# Phase 4 — per-DAG migration template
name: migrated-<old-dag-name>
prefect-version: 3.0.0
deployments:
  - name: migrated_<old_dag_name>
    entrypoint: flows/migrated/<old_dag_name>.py:<flow_function>
    parameters: <same as Airflow default_args>
    work_pool: { name: k8s-prod }
    schedules: [{ cron: "<same as Airflow schedule>" }]

Step-by-step explanation.

1. Phase 1 (weeks 1-4) is the pilot: pick one low-risk DAG (an internal-facing report is ideal), rewrite as a Prefect flow, ship in parallel with the Airflow version, compare outputs for a week, cut over. The purpose is proving out the operational story — SRE runbooks, on-call flow, observability.
2. Phase 2 (weeks 5-8) is the bridge: bidirectional REST-API calls between Airflow and Prefect. New Airflow DAGs can trigger Prefect flows; new Prefect automations can trigger Airflow DAGs. This lets subsequent migrations happen without a big-bang.
3. Phase 3 (weeks 9-16) is the freeze: no new Airflow DAGs; all net-new flows land on Prefect. This grows the Prefect footprint while the Airflow footprint plateaus.
4. Phase 4 (weeks 17-24) is the actual migration: 20 DAGs, ~1 day each, ordered by (a) risk (low-risk first), (b) coupling (isolated DAGs first, cross-DAG chains last), (c) business impact (internal-only before customer-facing). One migration per day-plus-review cycle.
5. Phase 5 (weeks 25-26) is retirement: verify no Airflow DAGs are running for a week, then shut down the Airflow deployment. Keep the Airflow MWAA / VM around for another month as a safety net; then destroy.

Output.

Month	Airflow DAGs	Prefect flows	New development lands on
0	20	0	Airflow (baseline)
2	20	1 (pilot)	Airflow + Prefect
4	20	5 (net-new)	Prefect only
6	0	25 (20 migrated + 5 new)	Prefect only

Rule of thumb. Never big-bang an orchestrator migration. Gradual coexistence via bridges — Airflow calling Prefect, Prefect calling Airflow — is the pattern that keeps risk bounded. Budget ~1 day per DAG for migration; add a 30% buffer for the last few (the trickiest are always at the end).

Senior interview question on orchestrator choice

A senior interviewer might ask: "A team of 5 data engineers at a Series B startup asks you to pick an orchestrator for a greenfield deployment. They have batch ETL (10 tables), streaming triggers (Kafka), and an ML feature backfill workload. Justify your recommendation across the four axes, cover the Prefect Cloud vs self-hosted decision, and outline the first-90-days migration plan."

Solution Using Prefect 3 Cloud, k8s work pool, bridge patterns for later Airflow legacy

# 1. First flow — the batch ETL for one of 10 tables
from prefect import flow, task
from prefect.tasks import task_input_hash
from datetime import timedelta

@task(retries=3, retry_delay_seconds=[10, 30, 60],
      cache_key_fn=task_input_hash, cache_expiration=timedelta(hours=24))
def extract(table: str, run_date: str) -> list[dict]:
    ...

@task
def load(rows: list[dict], target: str) -> int:
    ...

@flow(name="batch_etl_one_table")
def batch_etl_one_table(table: str, run_date: str):
    rows = extract(table, run_date)
    return load(rows, f"ANALYTICS.{table.upper()}")

# 2. prefect.yaml — three deployments (batch, streaming, ML) sharing one k8s work pool
name: startup-etl
prefect-version: 3.0.0
deployments:
  - name: batch_etl_hourly
    entrypoint: flows/batch.py:batch_etl_all_tables
    work_pool: { name: k8s-prod, job_variables: { image: registry.internal/etl:{{ get_current_commit_sha() }}, cpu: "500m", memory: "1Gi" } }
    schedules: [{ cron: "0 * * * *" }]

  - name: streaming_trigger
    entrypoint: flows/streaming.py:process_kafka_event
    work_pool: { name: k8s-prod, job_variables: { image: registry.internal/etl:{{ get_current_commit_sha() }} } }
    triggers:
      - type: event
        match: { prefect.resource.id: "prefect.event.kafka.event-received" }
        parameters: { event_id: "{{ event.payload.event_id }}" }

  - name: ml_feature_backfill
    entrypoint: flows/ml.py:backfill_features
    work_pool: { name: k8s-prod, job_variables: { image: registry.internal/etl:{{ get_current_commit_sha() }}, cpu: "2000m", memory: "8Gi" } }
    triggers:
      - type: event
        match: { prefect.resource.id: "prefect.event.dbt.completed" }
        parameters: { models: "{{ event.payload.models }}" }

# 3. Baseline automations — SLA + failure notification
- name: sla_batch_09_utc
  trigger: { type: event, posture: proactive, expect: [prefect.flow-run.Completed], within: 32400 }
  actions: [{ type: send-notification, block_document_id: <slack-oncall>, subject: "SLA breach" }]

- name: flap_suppressed_failure
  trigger: { type: event, posture: reactive, expect: [prefect.flow-run.Failed], within: 900, threshold: 3 }
  actions: [{ type: send-notification, block_document_id: <slack-oncall>, subject: "3+ failures in 15m" }]

Step-by-step trace.

Layer	Choice	Rationale
Orchestrator	Prefect 3 (Cloud)	dynamic + eventful + small team
Compute	k8s work pool	pay-per-run, autoscale
Deployments	3 (batch, streaming, ML)	one flow per workload
Triggers	cron for batch; event for streaming + ML	eventful-first design
SLAs	proactive automations	first-class deadline enforcement
Notifications	Slack + PagerDuty via blocks	severity-tiered
First-90-days plan	ship batch first; add streaming month 2; ML month 3	incremental risk

After the 90-day plan, the startup has three production deployments running on one shared k8s work pool with per-flow-run pod isolation, an events-driven bridge between Kafka and Prefect, an automations layer for SLAs and alerts, and a per-repo prefect.yaml that CI applies on merge. No Airflow footprint; no self-hosted control plane to maintain; ~$100/mo on Prefect Cloud starter tier.

Output:

Metric	Value
Orchestrator	Prefect 3 Cloud
Work pools	1 (k8s-prod)
Deployments	3 (batch, streaming, ML)
Baseline automations	2 (SLA + flap suppression)
Cloud spend	~$100/mo starter
SRE headcount	0 (Cloud manages control plane)
Ship-to-first-run time	Day 1

Why this works — concept by concept:

• Prefect 3 Cloud + k8s work pool — the Cloud tier removes control-plane ownership (no Postgres, no API server to run). The k8s work pool gives per-run pod isolation with cluster autoscaling — zero idle worker cost, full isolation per flow run.
• One flow per workload, one deployment per workload — clear ownership; each flow's SLA, retries, and compute profile are localised to its deployment. No shared-worker contention between batch ETL and ML backfill.
• Event-driven design from day one — the streaming and ML flows have event triggers rather than cron schedules. This future-proofs the design: adding another trigger source (S3 event, webhook, another Kafka topic) is one more automation, not a redesign.
• Baseline automations — SLA (proactive) + flap-suppressed failure (reactive + threshold). Two automations cover the "did the pipeline run?" and "is anything broken?" questions from day one, without inventing custom monitoring.
• Cost — ~$100/mo Cloud starter, existing k8s cluster (no new infrastructure), 0 SRE headcount for control plane. Compared to self-hosted Airflow (Postgres + scheduler + webserver + worker + on-call), Prefect Cloud + one work pool ships the same capability at a fraction of the ops load. O(1) infrastructure per workload class.

Design
Topic — design
Design problems on orchestrator selection

Practice →

Aggregation
Topic — aggregation
Aggregation problems on pipeline metrics

Practice →

---

Cheat sheet — Prefect 3 recipes

• The four primitives you must name in every Prefect answer. @flow decorates the top-level Python function; @task decorates its subroutines; a deployment versions a flow + parameters + schedule + triggers against a work pool; a work pool is a named compute abstraction (process, docker, kubernetes, ecs, cloud-run, vertex, serverless) that workers poll for runs. Every senior Prefect answer starts by naming these four; skipping any of them signals unfamiliarity. In Prefect 3, agents are gone; only workers pull from work pools — this is the single biggest 2 → 3 change.
• @flow / @task decorator template. @flow(name="...", log_prints=True, timeout_seconds=3600) on the entrypoint; @task(retries=3, retry_delay_seconds=[10, 30, 60], cache_key_fn=task_input_hash, cache_expiration=timedelta(hours=6), persist_result=True, result_storage=S3Bucket.load("prefect-results"), tags=["extract"]) on each subroutine. Retries as a list of intervals beats scalar for real workloads; cache_key_fn=task_input_hash gives free idempotent backfills; persist_result=True + a storage block gives worker-crash recovery. Skipping any of these three ships a fragile task.
• Retry + cache key template. For any task calling an external system: @task(retries=3, retry_delay_seconds=[10, 30, 60], retry_condition_fn=lambda t, r, s: isinstance(s.data, HTTPError)) — retries only on HTTPError, backs off aggressively. For any deterministic transform: @task(cache_key_fn=task_input_hash, cache_expiration=timedelta(hours=24)) — cache by input hash for 24h; identical inputs within the window return cached results without re-running. Cache invalidation on new code: bump a transform_version argument to force a fresh key.
• Result persistence + storage blocks. S3Bucket(bucket_name="prefect-results", aws_access_key_id=..., aws_secret_access_key=...).save("prefect-results") — one-time registration; then @task(persist_result=True, result_storage=S3Bucket.load("prefect-results")). On worker crash, downstream tasks fetch the persisted upstream result via the run UUID and skip re-computation. For interoperability across languages, use JSONSerializer instead of pickle; for pandas DataFrames, register a custom PandasSerializer writing Parquet.
• Deployment via prefect.yaml. Repo-root prefect.yaml declares all deployments: name:, entrypoint: flows/<file>.py:<flow_function>, parameters:, work_pool: { name: <pool>, job_variables: { image: <tag>, cpu: <req>, memory: <req> } }, schedules: [{ cron: "..." }], triggers: [{ type: event, match: {...}, parameters: {...} }]. Build/push image once in build: / push: blocks; tag with {{ get_current_commit_sha() }} for git-reproducible rollback. Ship dev + staging + prod deployments in the same yaml from day one — three entries pointing at the same flow entrypoint with different work pools.
• Work pool creation cheat sheet. prefect work-pool create dev-process --type process (dev laptop); prefect work-pool create staging-docker --type docker (single host); prefect work-pool create k8s-prod --type kubernetes (cluster autoscaled Jobs); prefect work-pool create ecs-serverless --type ecs-task (managed serverless); prefect work-pool create cloud-run --type cloud-run-v2 (GCP serverless). Start the worker inside the target compute: prefect worker start --pool k8s-prod inside the cluster; prefect worker start --pool dev-process on the laptop. One worker deployment per pool; scale replicas by expected flow-run concurrency.
• Kubernetes work-pool base template essentials. backoffLimit: 0 (Prefect owns retries, not k8s); ttlSecondsAfterFinished: 3600 (GC finished Jobs); tolerations + nodeSelector pinning to a dedicated prefect-runs node pool (isolate from other workloads); serviceAccountName: prefect-flow-sa (IAM binding for warehouse access via IRSA / Workload Identity); Jinja templating of image, cpu, memory from job_variables: so per-deployment resource overrides work without pool duplication. Sizes: base 500m CPU / 1Gi memory; deployment-level override for heavier flows.
• Automation trigger template. trigger: { type: event, posture: reactive, match: { prefect.resource.id: "prefect.deployment.<uuid>" }, expect: ["prefect.flow-run.Failed"], within: 900s, threshold: 3 } — fires on 3 failures of a specific deployment within 15 minutes. Change posture: reactive to posture: proactive and swap expect: to [prefect.flow-run.Completed] for SLA-style deadline enforcement (fires when the event does not arrive in the window). Compose with type: compound, require: any/all for AND/OR conditions across multiple triggers.
• Custom event bridge pattern. External signal (Kafka, S3 event, webhook, cron) → small Python bridge → PrefectEventsClient().emit(event="<domain>.<signal>", resource={"prefect.resource.id": "prefect.event.<domain>-<signal>"}, payload={...}) → automation with match: { prefect.resource.id: "prefect.event.<domain>-<signal>" } + run-deployment action templating parameters from {{ event.payload.<field> }}. This is the eventful-orchestration primitive; every external trigger source becomes a first-class Prefect input via ~20 lines of bridge code.
• Prefect vs Airflow decision matrix. Dynamic DAGs: Prefect 5, Airflow 3 (task mapping constrained), Dagster 4. Eventful triggers: Prefect 5 (automations), Airflow 3 (Datasets + sensors), Dagster 4 (sensors). Asset lineage: Dagster 5 (native SDA), Airflow 2, Prefect 2. Operator ecosystem: Airflow 5, Prefect 3, Dagster 3. Migration from Airflow: Airflow 5 (stay), Prefect 2 (rewrite), Dagster 2 (rewrite). Total: Airflow 19, Dagster 18, Prefect 17 — no single winner. Sweet spots: Prefect for greenfield dynamic + eventful; Airflow for legacy at scale; Dagster for asset-first lakehouse.
• Prefect 2 → 3 migration checklist. (1) Replace agents with workers: prefect worker start --pool <name>. (2) Replace Deployment.build_from_flow(...) with flow.deploy(...) or prefect.yaml. (3) Verify async flows are marked async def; sync flows are the default in 3. (4) Rewrite notification blocks as automations. (5) Migrate storage / result blocks — same block types, slightly different config schema. (6) Re-register blocks in the new workspace before switching over. Budget ~2 weeks for a mid-sized migration; longer if you have custom subclasses.
• Prefect Cloud vs self-hosted decision cheat sheet. Cloud: pay-per-user + per-run tier, hosted control plane, UI + events + automations included, ~$40/mo starter → thousands/mo at scale. Self-hosted: free OSS, run prefect server (Postgres + API + UI), own the on-call, ~1-2 SRE-days/month steady-state. Pick Cloud when: SRE headcount is scarce, compliance allows control-plane egress, or you're under 100k flow runs/month. Pick self-hosted when: air-gapped, cost sensitivity dominates, or you already have an SRE team that runs Postgres well.
• Observability essentials. Every production Prefect workspace needs: (1) baseline SLA automation per critical deployment (proactive posture, Slack for warnings, PagerDuty for pages), (2) flap-suppressed failure automation (threshold 3 within 15m → Slack), (3) worker-offline automation (worker down > 5 min → PagerDuty), (4) tag-based flow filtering (tags=["prod", "customer-facing"] for cross-workspace grouping), (5) structured logs via log_prints=True on @flow for print-capture. Ship all five on day one; retrofitting them under a fire is painful.
• When to use subflows vs tasks. Subflow when: (a) the block has its own retry / caching / result policy distinct from the parent, (b) the block is reused across multiple parent flows, (c) the block deserves independent observability in the UI. Task otherwise. Do not subflow just for visual grouping — tags cover that. Subflow overhead is one extra flow-run row per call; noticeable in tight inner loops (thousands of subflow calls per parent). For fan-out over a list, prefer task mapping via .map() unless each element needs subflow-level policy.
• Cost budgeting for Prefect deployments. Prefect Cloud starter: ~$40/mo (small team, low run volume). k8s work pool cost: ~$5/mo for one worker Deployment (idle); flow-run pods cost only during execution (auto-scaled node pool). Custom bridges (Kafka, S3, webhook): 1 pod each, ~$5/mo. Total for a small-team production setup: ~$100/mo control plane + variable per-run compute. Compared to a self-hosted Airflow deployment (Postgres + scheduler + webserver + worker + always-on nodes), Prefect's serverless-first work pools are dramatically cheaper below ~1M runs/month.

Frequently asked questions

What is Prefect 3 in one sentence?

prefect 3 is a Python-first orchestrator whose four load-bearing primitives — @flow and @task decorators for imperative Python pipelines, deployment objects versioning a flow against a work pool compute abstraction, and automations reacting to typed events with compound-condition triggers — collapse the four things Airflow makes hard (dynamic DAGs, sub-second triggers, per-flow compute isolation, and reactive alerting) into ordinary Python plus a small prefect.yaml file, delivered as either a hosted Prefect Cloud control plane or a self-hostable OSS server. The 2024 rewrite from Prefect 2 to Prefect 3 removed agents (only workers now), defaulted to sync execution (async is opt-in), and elevated events + automations to first-class concepts, which is why every senior data-engineering interview in 2026 that touches orchestration asks about Prefect 3 by name — the primitives, the deployment model, and the eventful-triggers story are the shape that most differentiates it from Airflow's DAG-centric worldview and Dagster's asset-centric worldview.

Prefect 3 vs Airflow — when do I pick each?

Default to Prefect 3 when your workload is dynamic (runtime-known fan-out over lists that aren't visible at parse time), eventful (sub-second reactive triggers on custom conditions like Kafka messages, S3 events, or dbt-completed signals), or needs per-flow compute isolation (one k8s Job per run, different CPU/memory profiles per deployment, or a mix of process/Docker/Kubernetes/serverless pools). Prefect's automations engine + typed events + compound-condition triggers is the biggest capability advantage; the work-pool abstraction is the biggest ergonomic advantage. Default to Airflow 3 when you already have a mature Airflow deployment (hundreds of DAGs, invested SREs, working CI/CD), your integrations are dominated by Airflow-only operators, or migration cost isn't justified by a workload shape that Prefect does better. Both handle static hourly ETL comfortably; the choice at that shape is about developer ergonomics (Prefect's edge) vs the operator ecosystem depth (Airflow's edge). Never pick an orchestrator based on trend — pick based on (dynamic × eventful × compute abstraction × migration cost), and if you're greenfield in 2026, the answer is Prefect more often than not.

What is a work pool and why did Prefect remove agents?

A prefect work pool is a named compute abstraction — one of process, docker, kubernetes, ecs-task, cloud-run-v2, vertex-ai, or push-to-serverless — that carries a base job template (image, CPU/memory requests, service account, tolerations) and a queue of pending flow runs from deployments that name the pool. A worker is a long-running process (prefect worker start --pool <name>) that polls the pool, pulls scheduled flow runs, and provisions infrastructure per run according to the base template and per-deployment job_variables: overrides. Prefect 2 had both agents (per-queue pullers) and workers (compute drivers), which created a confusing "which one do I run?" surface — Prefect 3 collapsed the two into a single primitive (workers pull from work pools), giving one mental model and one operational primitive. The work pool also decouples "what runs" (the flow) from "where it runs" (the pool), so one flow can have a dev deployment on a process pool, staging on docker, and prod on kubernetes — all from one prefect.yaml file — a shape that Airflow's install-time-fixed executor cannot cleanly express.

How do deployments differ from flows?

A flow is the Python code — the @flow-decorated function plus its @task-decorated subroutines. A deployment is a durable, versioned binding of (a flow entrypoint + parameters + schedule + triggers + work pool + job variables) stored in Prefect's API and applied via prefect deploy. One flow can back many deployments (dev / staging / prod, or different regions, or different parameter defaults). The flow doesn't know it's being deployed multiple ways; the deployment layer is entirely orthogonal to the flow definition. This split is Prefect's biggest ergonomic win over Airflow, where "the DAG file" bundles the code, the schedule, and the deployment metadata into one Python module. Deployments are versioned — each prefect deploy creates or updates a deployment with a version string (default: git SHA), so rollback is git checkout <old-sha> && prefect deploy in under 30 seconds. Schedules and triggers live on the deployment, not the flow: adding a new schedule = adding a new deployment; changing a trigger = editing one line in prefect.yaml.

How do automations differ from schedules?

A schedule fires a flow run at pre-defined times (cron, interval, or rrule); it is pull-based — the scheduler wakes up periodically and starts runs whose time has arrived. An automation fires an action in response to a typed event matched against a compound-condition trigger; it is push-based — the event bus receives events (flow-run state transitions, custom emissions via emit_event, block updates, worker lifecycle) and the automations engine matches them against configured rules. Actions include send-notification (Slack, email, PagerDuty via blocks), run-deployment (start another flow), pause-deployment, cancel-flow-run, send-event (chain to another automation), and send-webhook. This is Prefect 3's biggest new capability and the shape that pushes the orchestrator beyond scheduling into reactive orchestration — SLA enforcement (posture: proactive fires on the absence of a Completed event within a window), flap-suppressed failure notification (threshold: 3 within: 15m), and cross-flow triggers (event from flow A starts a run of flow B) all become one-page automation configs instead of custom monitoring stacks. Airflow's closest analogues (Datasets, ExternalTaskSensor, on_failure_callback) sit lower in the stack and require more glue.

Is Prefect Cloud required for production?

No — prefect 3 ships fully open-source, and you can run the entire control plane yourself via prefect server start backed by Postgres. Prefect Cloud is a hosted, multi-tenant version of that same control plane, with additional features (SSO, RBAC, audit logs, higher automation throughput, longer event retention) that scale with tier. Pick Prefect Cloud when your SRE headcount is scarce, your compliance regime allows control-plane egress (you install a worker in your VPC; the worker calls out to Cloud but never sends flow-run data), or you're under ~1M flow runs/month where the ~$40/mo starter tier undercuts self-hosting SRE cost. Pick self-hosted OSS when air-gapped compliance forbids control-plane calls, you already have an SRE team that runs Postgres and ASGI apps competently, or you're at multi-million-run scale where Cloud pricing starts to sting. Migration between Cloud and self-hosted is portable — the flow / deployment / work pool / automation concepts are unchanged; you swap the PREFECT_API_URL env variable, re-register blocks, and re-apply deployments. Most teams start on Cloud (lower time-to-first-flow) and migrate to self-hosted later if scale or compliance demands it.

Practice on PipeCode

• Drill the SQL practice library → for the incremental-load, idempotent-backfill, and pipeline-correctness problems that Prefect flows and cache keys are built to solve.
• Rehearse on the design practice library → for the orchestrator-selection, multi-environment deployment, and SLA-alerting design problems senior interviewers love.
• Sharpen the streaming axis with the streaming practice library → for event-driven triggers, Kafka-to-orchestrator bridges, and reactive pipeline patterns.
• Practice the aggregation library → for pipeline-metrics rollups — the shape most Prefect observability dashboards ultimately query.
• Stack the prerequisites against PipeCode's broader 450+ data-engineering catalogue to anchor the flow / deployment / work-pool / automation vocabulary against real graded inputs.

Lock in prefect muscle memory

Docs explain primitives. PipeCode drills explain the decision — when a subflow earns its own retries, when a work pool split pays for itself, when an automation replaces a whole monitoring stack, when Prefect Cloud beats self-hosted. Pipecode.ai is Leetcode for Data Engineering — pattern-first practice tuned for the orchestration trade-offs senior data engineers actually face.

Practice design problems →
Practice streaming problems →