Design Apache Airflow

• Define pipelines as DAGs in code: Python files in a repository declare tasks and dependencies. Code, so pipelines can be generated, reviewed and tested; a graph, so the scheduler knows what can run in parallel.
• Schedule runs per data interval: A daily DAG produces one run per day, and that run is about one specific day of data (its data interval), not about whatever “now” is when it happens to execute.
• Run tasks when their upstreams succeed: Plus trigger rules for the exceptions: run on failure, run when all upstreams are done whatever their result, skip branches.
• Retry, alert, and time out: Per-task retries with a delay, an execution timeout, and alerts or callbacks when a task fails for good or a run misses its deadline.
• Wait for external data: Sensors that wait for an upstream partition, file or API state, sometimes for hours, without wasting compute while they wait.
• Backfill and rerun: Rerun a failed task, clear and rerun a range of days after a bug fix, or run a new DAG for the past year, safely and at a controlled pace.
• Limit pressure on shared systems: Pools and concurrency limits, so a thousand tasks ready at midnight do not open a thousand connections to the warehouse.
• Out of scope: The compute engines themselves (Spark, the warehouse), streaming pipelines, data lineage and catalogue, and the DAG authoring IDE.

• Scale (20 k DAGs · ~1 M task runs/day): Throughput is modest. The difficulty is the burst at midnight UTC and keeping one coordination database healthy.
• Correct data per interval (rerunning a task produces the same result): The requirement that drives the hardest trade-off: retries, zombies and backfills all rerun tasks, so tasks must be idempotent per data interval. The scheduler can guarantee a task is retried, not that it ran exactly once.
• Scheduling latency (task queued < 10 s after its upstreams finish): A 30-task pipeline with a minute of overhead per hop wastes half an hour every night.
• Scheduler availability (no single point of failure): If the scheduler stops, nothing new starts: every pipeline in the company stalls. Running tasks are unaffected, which buys minutes, not hours.
• Isolation from user code (a broken DAG file breaks only itself): DAG files are arbitrary Python written by hundreds of people. A slow import or an infinite loop at parse time must not stall scheduling for everyone else.
• Auditability (every task try kept 1 year): Who ran what, for which interval, with which logs. Data teams debug last month’s numbers.

• Task runs per day: ~1 M. 20 k DAGs × ~2 runs a day on average (mostly daily, some hourly) × ~25 tasks per DAG = 1 M task instances a day, about 12/s on average. Throughput is not the hard part.
• Midnight burst: ~400 tasks/s. About 40 % of daily DAGs are scheduled for 00:00 UTC: 8 k DAG runs created in the first minute, each with ~3 root tasks, so 24 k task instances become ready within a minute ≈ 400/s, about 30× the average.
• Concurrent tasks: ~7 k avg. Average task ~10 minutes: 12 task starts/s × 600 s = ~7 200 running at once by Little’s law. After midnight the burst pushes this to two or three times as many, which the queue absorbs while workers scale.
• DAG parsing cost: ~34 cores. 5 k DAG files (many generate several DAGs) re-parsed every 30 s = ~170 parses/s; at ~200 ms of CPU each that is ~34 cores doing nothing but importing Python. This is why parsing cannot live inside the scheduler loop.
• Metadata rows per year: ~365 M task rows. 1 M task instances a day × 365 = 365 M rows a year in the task table alone, at ~2 KB each with rendered fields ≈ ~730 GB. The scheduler’s queries scan recent rows, so old rows must be archived or the hot indexes rot.
• Slots wasted by poking sensors: ~3 k. ~3 k sensors waiting at any time, averaging 2 hours each. In poke mode each holds a worker slot while it sleeps: 3 k slots, ~40 % of average capacity, doing nothing. One async triggerer process can watch thousands.

• DAG repository (Git, synced to hosts): Pipelines as Python files, reviewed and merged like any code. Synced to the DAG processors and the workers, so the code that is parsed is the code that runs.
• Data engineers: Write and merge DAG files, watch runs in the UI, clear failed tasks and launch backfills. Hundreds of them, of varying experience: the system must survive their mistakes.
• DAG processor (isolated processes): Imports each DAG file in a separate, time-limited process and writes the resulting graph, as JSON, to the metadata database. Keeps arbitrary user Python (slow imports, network calls, infinite loops) out of the scheduler entirely.
• Web UI & API: Grid and graph views, logs, manual triggers, clearing tasks and backfills. Reads the serialised DAGs and run state from the database; never imports DAG files itself.
• Schedulers (active-active, ×N): In a loop: create due DAG runs, find task instances whose upstreams are satisfied, apply pools and concurrency limits, and hand runnable tasks to the executor. Several run at once; row locks with SKIP LOCKED stop them stepping on each other.
• Metadata DB (Postgres): The single source of truth: serialised DAGs, DAG runs with their data intervals, every task instance and its state, pools, connections metadata, triggers. Every component coordinates through it, which makes it the system’s real bottleneck.
• Alerting: Failure callbacks, missed deadlines and retries-exhausted notifications to Slack, email or a pager, routed to the owning team named in the DAG.
• Executor queue (Celery broker / K8s): Decouples deciding a task can run from running it. With a queue-based executor, tasks wait here for a free worker slot; with a Kubernetes executor, each task becomes a pod request.
• Triggerer (asyncio, thousands per process): Runs deferred waits (is the partition there yet? has the job finished?) as lightweight async coroutines. When a condition is met it marks the waiting task to resume. Thousands of waits per process instead of one worker slot each.
• Workers (autoscaled slots): Load the DAG file, render the task’s templates for its data interval, fetch credentials, and run it, heartbeating while it runs. Most tasks are thin: they tell the warehouse or Spark what to do and wait.
• Secrets backend (Vault / cloud KMS): Connection credentials resolved at task run time, so passwords are not stored in DAG files or in the metadata database in plain text.
• Task logs (object storage): One log per task try, shipped to object storage when the try ends and streamed live while it runs. The UI reads them from here, so a dead worker’s logs are not lost with it.
• Warehouse & Spark: Where the data work actually happens. The orchestrator submits queries and jobs; it does not move data through itself. Pools protect it from being flooded at midnight.
• Data lake (partitioned by date): Tables partitioned by date. Each run writes exactly the partition for its data interval, by overwrite, which is what makes reruns and backfills safe.

1. A nightly pipeline runs for yesterday: The core path: a run created for a specific data interval, tasks released as their upstreams finish, and each task writing exactly its slice of data.
   1. A DAG processor parses the merged DAG file and stores its serialised graph.
   2. Just after midnight, a scheduler sees the DAG is due and creates a run for yesterday’s interval.
   3. The scheduler marks the root tasks scheduled and queues them, within pool limits.
   4. A worker picks up a task, loads the DAG file and renders its templates for the interval.
   5. The task submits a query that overwrites that day’s partition, then reports success.
   6. The next scheduler loop sees the upstream succeeded and releases the downstream tasks.
2. Wait for upstream data without holding a slot: A task that may wait hours for another team’s partition, without occupying a worker the whole time.
   1. A sensor task starts on a worker and checks whether the upstream partition exists yet.
   2. It defers: the worker writes a trigger record and frees its slot.
   3. A triggerer claims the trigger and checks for the partition every five minutes, asynchronously.
   4. The partition appears; the triggerer fires an event and the task is scheduled to resume.
   5. A worker resumes the task, which finishes in seconds, and downstream tasks are released.
3. Midnight UTC: eight thousand runs at once: The scale case: tens of thousands of tasks ready in a minute, several schedulers working at once, and a warehouse that must not be flooded.
   1. At 00:00 three schedulers each look for due DAGs and create runs, skipping rows another has locked.
   2. About 24 k root tasks become schedulable within a minute.
   3. Pools cap how many tasks may hit the warehouse at once; the rest stay scheduled.
   4. Queued tasks wait in the executor queue while workers scale out on its depth.
   5. Runs that miss their deadline page their owners.
4. A worker dies halfway through a task: The failure path: the task is detected as a zombie, retried, and safe to retry only because it overwrites its partition.
   1. A worker running a two-hour task is killed: a node is reclaimed, or the process runs out of memory.
   2. A scheduler notices the task’s heartbeat is older than the threshold and marks it failed.
   3. The task has retries left, so it is scheduled again after its retry delay.
   4. The second try runs on another worker and overwrites the same partition.
   5. If retries run out, the failure is alerted and downstream tasks wait as upstream_failed.
5. Backfill ninety days after a bug fix: The rerun path: correcting three months of output in a controlled way, oldest first, without starving tonight’s runs.
   1. An engineer merges a fix and clears the task for the last ninety intervals.
   2. Schedulers rerun the cleared tasks, at most a few runs of this DAG at a time.
   3. Each rerun overwrites its own day’s partition with corrected data.
   4. The engineer follows progress and logs in the grid view.

Ask: Why not have the scheduler import the DAG files directly? It needs the graph anyway.

Good answers name: A separate DAG processor parses files in time-limited child processes and stores serialised graphs in the database, The scheduler parses files inline in its loop (early Airflow), Declarative pipeline definitions (YAML or JSON) instead of code.

Our pick: Run DAG processors as their own service: each file is imported in a child process with a timeout (say 60 s), and the resulting graph is stored as JSON with a hash, skipping the write when nothing changed. Import errors are recorded per file and shown in the UI and to the owning team. The scheduler and webserver read only the serialised DAGs, so neither ever executes user code. Workers import the specific file when they run one of its tasks, since the task’s Python must run, and a version id on the serialised DAG lets the run record which version it started with. Airflow made serialised DAGs the default in 2.0 and the standalone processor mandatory in 3.0, for exactly these reasons.

1. An author merges a DAG that takes 90 seconds to import. What happens?
   Its child process is killed at the 60-second timeout and an import error appears for that file. Its previous serialised version, if any, keeps scheduling, so yesterday’s pipeline still runs. No other DAG is delayed, because each file parses in its own process.
2. A task is removed from a DAG while a run that includes it is in progress. What should happen?
   The run was created from a specific serialised version, so it can either finish with its original task list or mark the removed task as removed and skip it. The important part is that it is a defined behaviour recorded on the run, not whatever the latest parse happens to say mid-run.
3. How would you cut the 34 cores of parsing?
   Parse on change rather than on a timer where possible (watch the repository), raise the interval for files that rarely change, and keep top-level code cheap: no network calls or heavy imports at module level, with a lint rule in CI that measures import time. Most of the cost is avoidable.
4. How do you stop one team’s broken DAG from blocking a deploy for everyone?
   Validate in CI before merge: import every changed file in a clean environment with the same timeout, check for cycles and duplicate ids, and fail the merge if it breaks. The processor’s isolation is the safety net; CI is the first line.

Ask: Why does a daily run need a “logical date”? Why not just run the job every night at midnight?

Good answers name: Each run owns a data interval; tasks read the interval from the run and overwrite exactly that slice, Cron-style: run at a wall-clock time, tasks use “now”, Data-aware triggering: run when the upstream dataset is updated.

Our pick: Every DAG run carries data_interval_start and data_interval_end, created by the schedule (a daily DAG’s run covers one day and is created once that day has ended). Templates expose the interval ({{ ds }}, {{ data_interval_start }}), and tasks write output keyed by it with overwrite semantics: INSERT OVERWRITE a date partition, or delete-then-insert for the interval in one transaction. With catchup on, a DAG whose start date is in the past gets one run per missed interval, paced by max_active_runs. The orchestrator guarantees each interval’s task is attempted until it succeeds or exhausts retries; the task’s idempotency is what makes repeated attempts produce one correct result.

1. A task appends rows to a table instead of overwriting a partition. What goes wrong, and how do you fix it without partitions?
   Every retry and every rerun adds the interval’s rows again, so totals drift upward invisibly. Without partitions, wrap the write in one transaction that deletes rows for the interval and inserts the new ones, or MERGE on a natural key. Either makes a rerun replace rather than add.
2. Upstream data for a day arrives late and is corrected a week later. What should happen downstream?
   Clear and rerun the affected interval and its downstreams; overwrite semantics make that safe. With dataset events, the corrected upstream write can trigger the downstream reruns automatically. Either way, the unit of correction is the interval, which is the point of the model.
3. Should a run for today wait for yesterday’s run to succeed?
   Only if the task genuinely depends on its own previous output, such as a cumulative snapshot; then depends_on_past makes the order explicit. For most daily partitions, independent intervals are better: one bad day does not block the following ones, and backfills can run in parallel.
4. A DAG is deployed with a start date a year ago and catchup on. Is that a problem?
   It immediately has 365 runs to do. With max_active_runs = 1 and a low priority it trickles through safely; with defaults it can flood the pools at once. Platforms usually default catchup off for new DAGs and make large backfills an explicit, paced action.

Ask: The scheduler is a single point of failure. How do you run more than one without them scheduling the same task twice?

Good answers name: Active-active schedulers coordinating through row locks with SKIP LOCKED, One active scheduler with a leader election and a hot standby, Shard DAGs across schedulers by hash of dag_id.

Our pick: Run two or three schedulers, all active. Each loop creates due runs and examines schedulable task instances by selecting batches with FOR UPDATE SKIP LOCKED, so two schedulers never examine the same rows at the same moment and never wait on each other. State changes (scheduled to queued) commit in the same transaction, and unique constraints on runs and task tries are the backstop if anything slips through. The database is the coordinator, so it gets connection pooling (PgBouncer), indexes designed for the scheduler’s queries, and close monitoring. This is how Airflow 2 made its scheduler highly available without adding ZooKeeper or a consensus service.

1. Why is SKIP LOCKED essential? Couldn’t schedulers just use normal row locks?
   With normal locks, the second scheduler blocks until the first commits, so schedulers take turns and the extra ones add nothing. SKIP LOCKED lets each take the rows nobody holds, so they work on disjoint batches in parallel.
2. A scheduler crashes after marking a task queued but before sending it to the executor. What happens?
   The task sits in queued with no worker. A periodic check finds tasks queued longer than a timeout and resets them to scheduled, so another scheduler requeues them. The task may then run twice if the original message did reach the queue, which is one more reason tasks are idempotent.
3. How many schedulers is too many?
   When they start competing for the same database. Each loop runs several queries; past a handful of schedulers, the database spends its time on locks and connections rather than useful work. The signal is scheduler loop duration rising while CPU on the schedulers is low.
4. What does the database need to survive this?
   A connection pooler, because every scheduler, worker and webserver opens connections; indexes matching the hot queries (task state, DAG run state, heartbeat); routine archiving of old rows; and a replica for the UI’s heavy reads. It is the component to watch first when scheduling slows.

Ask: Should tasks run in a shared worker pool or each in its own container?

Good answers name: Thin tasks on a shared, autoscaled worker pool; heavy or dependency-conflicting tasks in their own pods, Every task in its own Kubernetes pod, Run all tasks inside the scheduler process.

Our pick: Default to a queue-based executor with an autoscaled pool of workers, each running many thin tasks that submit work to the warehouse or Spark and wait, heartbeating. Route tasks that need special dependencies, GPUs or large memory to a Kubernetes path where each runs in its own pod with explicit resource requests. Make the routing a field on the task (queue or executor) so authors choose deliberately. Platform guidance and lint rules push teams away from pulling large datasets into the worker, because that turns the orchestrator into an under-provisioned compute engine.

1. A task downloads a 50 GB file into the worker to process it. What is wrong with that?
   The worker becomes the compute engine: it fills its disk, starves its neighbours’ memory, and fails in ways that look like orchestrator bugs. The task should submit a Spark job or a warehouse query that processes the file where it lives, and wait for the result.
2. How do you size the worker pool?
   Average concurrency is about 7 k tasks by Little’s law, so run around that many slots during the day and autoscale on queue depth for the midnight burst. Because most tasks are waiting on external systems, slots are cheap: many per worker process.
3. Why not use the Kubernetes path for everything and be done with dependency conflicts?
   At 400 task starts a second after midnight, pod creation becomes the bottleneck, and adding 20 to 60 seconds of start-up to a 10-second task triples pipeline duration. Isolation per task is valuable for long, heavy tasks and wasteful for short ones.
4. A worker host is lost. What happens to its tasks?
   Their heartbeats stop, schedulers mark them as zombies after the threshold and retry them elsewhere. Logs streamed to object storage survive the host. Nothing needs a person, as long as the tasks are idempotent.

Ask: A sensor waits two hours for another team’s data. What does that wait cost you?

Good answers name: Deferrable operators: the task suspends into an async trigger run by a triggerer service, Poke mode: the sensor loops inside its worker slot, Reschedule mode: the sensor exits between checks and is rescheduled later.

Our pick: Sensors and operators that wait on external systems defer: on the first negative check they write a trigger (a small async class plus its arguments) to the database, move to a deferred state and release their worker slot. Triggerer processes claim triggers with the same skip-locked pattern, run each as a coroutine that checks every few minutes, and on success fire an event that moves the task back to scheduled with the result attached. A crashed triggerer’s triggers are re-claimed by another. Where the producing DAG is on the same platform, dataset events remove the need to wait at all: the consumer runs when the producer writes.

1. Why can poke-mode sensors deadlock a deployment?
   If waiting sensors occupy every worker slot, the upstream tasks that would produce their data cannot get a slot to run. Everyone waits for everyone, and nothing in the system detects it. Deferral makes this impossible because waiting no longer consumes the resource the producers need.
2. The triggerer itself crashes. Are the waits lost?
   No, triggers are rows in the database. Another triggerer notices the dead one’s heartbeat has stopped and claims its triggers, then resumes checking. A check might run twice, which is harmless for a read.
3. How do you handle data that never arrives?
   Every sensor has a timeout. When it expires the task fails, alerts its owner and stops downstream tasks running on missing data. Far better than a sensor that waits forever and a dashboard that silently shows yesterday.
4. Could the producer just tell us when the data is ready?
   Yes, and that is better where possible: a dataset event when the producer’s task writes the partition, or a message on a queue from an external system. Sensors are for producers that cannot or will not notify.

Ask: Everything coordinates through one Postgres. When does that break, and what do you do about it?

Good answers name: Keep one relational database: pool connections, index for hot queries, archive old rows, offload UI reads, Move state to a distributed NoSQL store, Split into several independent deployments, by team or domain.

Our pick: Put a connection pooler in front of the database, since every worker and scheduler opens connections; index for the scheduler’s queries (task state by run, heartbeat age, DAG next run time) and review slow queries after each upgrade. Archive task instances, logs references and runs older than the retention period (a cleanup job, for example 90 days hot, then exported), so the hot tables stay small and vacuum keeps up. Send UI history reads to a replica. Reduce chatter at the source: heartbeats every few seconds rather than every second, and batched state updates. When one deployment approaches its ceiling, split by domain rather than rebuild on a different store. Airflow 3 also moved workers off direct database access to an API server, which cuts connection counts and lets the platform control what tasks can touch.

1. Scheduling slows down over a few months with no change in load. What do you look at first?
   Table and index bloat from unarchived history: the scheduler’s queries are scanning far more rows than they did. Check the size of the task table and the query plans for the hot queries; an archiving job usually fixes it, and a missing index fixes the rest.
2. Why not just give the database a much bigger machine?
   It helps, once. Connection counts and lock contention grow with workers and schedulers, not just data size, and a bigger machine does not fix 365 M rows a year accumulating in hot indexes. Hygiene first, then size.
3. The database fails over to its replica. What happens to running tasks?
   Running tasks keep running in the warehouse or in their pods; their heartbeats and state updates fail briefly and retry. Schedulers reconnect and continue. After failover, any tasks that look like zombies because heartbeats were lost are retried, which is safe if they are idempotent.
4. How do you know when it is time to split the deployment?
   When scheduler loop time or database CPU stays high after the usual fixes, when one team’s backfills regularly delay others, or when teams cannot agree on an upgrade window. These are organisational signals as much as technical ones.