Airflow DAG Authoring

The reasoning

With TaskFlow you write `total = transform(extract())` and get three things for free: the tasks, the dependency edges, and the data passing. Without it you write two `PythonOperator`s, an explicit `extract >> transform`, an `xcom_push` or a return, and an `xcom_pull` with the right task id and key in the downstream callable — four places to keep in step instead of one expression.

It is also more testable. A `@task`-decorated function is still a plain function; `my_task.function(arg)` calls the undecorated body, so unit tests do not need an Airflow context at all. And type hints on the parameters are real hints, where `op_kwargs` is an untyped dict.

Classic operators are still right for everything that is not Python. `SQLExecuteQueryOperator`, `KubernetesPodOperator`, `S3ToRedshiftOperator` and the rest of the provider catalogue exist precisely so you do not write that code yourself, and reaching for `@task` with a hand-rolled boto3 call instead of the provider operator is the opposite mistake. The two mix freely in one DAG — a TaskFlow function can be `>>`'d to a classic operator.

The one thing to be careful about is that TaskFlow makes XCom invisible. `transform(extract())` looks like a function call and is actually a database round trip, so the habit of passing a pointer rather than a payload matters *more* in this style, not less — it is easier to accidentally return a DataFrame when it looks like ordinary Python.

The reasoning

Two separate problems. First, testability: 120 lines mixing Airflow context lookups with business logic can only be tested by running Airflow. Move the transformation into an ordinary function that takes and returns data, and the task becomes three lines that pytest never needs to see.

Second, granularity. One task doing extract, transform and load means a failure in the load retries the extract — which is wasted work at best, and at worst hits a rate-limited API again for data you already have. Splitting into tasks makes the retry boundary match the failure boundary, and the graph in the UI tells you which step failed instead of just 'the task'.

The counterweight, because splitting too far is also a mistake: every task boundary is an XCom round trip and a scheduling delay, so a hundred one-line tasks is slower and noisier than five meaningful ones. The rule I use is that a task should be the unit you would want to retry independently. If retrying half of it makes no sense, it is one task.

The specific fix here: read the Variable via a template or inside the task rather than at module level; make the warehouse query its own task that writes to a known location; make the transform a pure function called by a task; make the S3 write its own task keyed on the logical date so a rerun overwrites rather than duplicates.

The reasoning

`load.expand(name=list_files())` creates one task instance of `load` per element of the upstream's return value. They share a task id and are distinguished by `map_index`, and each gets its own state, logs and retry counter — so one bad file fails one instance rather than the whole fan-out.

The timing is the important part. At parse time Airflow knows the mapped task exists but not how wide it is; after `list_files` succeeds, the scheduler expands the mapped task into concrete instances. That is why this replaced the old pattern of building tasks in a `for` loop at parse time, which required the list to be known when the file was imported and produced a DAG whose shape changed under you.

A downstream task taking the mapped result receives the list of all return values — the reduce step — so `summarise(load.expand(...))` gets `["loaded a.csv", ...]`. If you want the downstream to also be mapped, you chain another `.expand()`.

The practical bits worth having ready: `partial()` supplies the arguments that do not vary — `load.partial(bucket='x').expand(name=...)`. `expand_kwargs()` maps over a list of dicts when several arguments vary together. `max_active_tis_per_dag` caps how many instances run at once, which matters because a list that unexpectedly returns ten thousand elements will happily try to schedule ten thousand tasks. And `max_map_length` is the hard ceiling that stops it.

The reasoning

A branch operator returns the task id (or ids) to follow; everything else directly downstream is marked `skipped`. That part surprises nobody. What surprises people is what happens next: `skipped` propagates. The default trigger rule is `all_success`, which requires *every* direct upstream to have succeeded, and a skipped task has not.

So `publish` is skipped, the run is reported successful, and the thing you actually wanted to publish never ran. Nothing is red. That silence is why this question is asked — a candidate who has hit it recognises it in seconds, and one who has not will insist the DAG is broken.

The fix is the trigger rule. `none_failed_min_one_success` is the usual choice for a join after a branch: it runs as long as nothing upstream failed and at least one succeeded, so a skipped sibling is fine. `none_failed` is the looser version that also runs when everything upstream was skipped, which is occasionally what you want and usually not.

The design alternative worth mentioning: if both branches genuinely converge, put the join immediately after each branch rather than after both, or use a TaskGroup per branch so the skip is contained. Reaching for `all_done` here is the wrong instinct — it makes `publish` run even when the branch it depended on *failed*, which trades a silent skip for a silent bad publish.

The reasoning

The only change is one string, and `publish` goes from `skipped` to `success`. That is the whole lesson: the graph was never wrong, the join condition was.

What the rule permits precisely: upstreams may be `success` or `skipped`, at least one must be `success`, and none may be `failed` or `upstream_failed`. That is exactly the semantics a post-branch join wants — tolerate the path not taken, still stop if the path that was taken broke.

The neighbouring rules, so you can pick deliberately. `all_success` is the default and the one that bites. `none_failed` is like the above but also runs when *everything* upstream skipped, which means a join that fires even though no branch ran. `all_done` runs regardless of upstream state and is for cleanup, not for joins. `one_success` fires as soon as any upstream succeeds without waiting for the others. `all_skipped` and `one_failed` exist for the reporting and alerting shapes.

The rule of thumb: use `all_success` everywhere by default, `none_failed_min_one_success` at a join after a branch, and `all_done` only for teardown that must happen either way. Any other choice deserves a comment saying why, because a non-default trigger rule is how a DAG ends up running on data it should not have.

The reasoning

`all_done` means 'run once every upstream has reached a terminal state, whatever that state is'. It is the correct rule for cleanup: dropping a temp table, releasing a lock, removing a scratch directory — things that must happen whether the pipeline worked or not.

`load` gets `upstream_failed`, which is a distinct state from `failed`. It never executed; Airflow marked it as unable to run. That distinction matters in an incident, because the tasks you need to investigate are the `failed` ones and the `upstream_failed` ones are just consequences.

The run is `failed` regardless of the cleanup succeeding, and that is the right behaviour — a successful teardown does not mean the pipeline delivered. Anyone relying on run state for alerting gets told, which is what you want.

Two cautions on `all_done`. It is not a general-purpose 'be permissive' rule: putting it on a real processing task means that task runs on the output of something that failed, which is how corrupt data gets published. And a cleanup task that itself fails will fail the run even when everything else worked, so cleanup should be defensive — drop-if-exists rather than drop.

The reasoning

`a >> b` sets a single edge. `a >> [b, c]` fans out. `[a, b] >> c` fans in. `[a, b] >> [c, d]` is the one to be careful about — it creates every combination, so two lists of ten give you a hundred dependency edges. That is often what you want semantically but it makes the graph unreadable and adds real scheduler work.

`chain()` from `airflow.sdk` (previously `airflow.models.baseoperator`) handles the common shapes: `chain(a, b, c)` is a linear sequence, and `chain([a, b], [c, d])` pairs them up — a→c and b→d — rather than crossing them. `cross_downstream([a, b], [c, d])` is the explicit way to ask for the cartesian version, and being explicit is the point.

For real fan-in, a single join task is usually better than a hundred edges: have the ten first-stage tasks feed one `EmptyOperator` and the ten second-stage tasks depend on that. The graph goes from a hundred edges to twenty, it renders legibly, and the semantics are identical.

TaskGroups are the other half of this. Grouping the two stages collapses them in the UI and lets you set dependencies between groups, which is how a fifty-task DAG stays comprehensible. They are purely an authoring and display construct — no runtime cost, unlike the SubDAGs they replaced, which ran their own scheduler and were removed in Airflow 3.

The reasoning

A `TaskGroup` wraps tasks into a collapsible unit in the UI and namespaces their task ids as `group.task`. Dependencies can be set on the group as a whole, so `extract_group >> transform_group` wires every edge you meant. At runtime there is nothing there — the tasks are ordinary tasks in the same dag run, scheduled exactly as if the group did not exist.

SubDAGs were the opposite: a `SubDagOperator` was a task that created and ran a whole nested dag run. That meant a task occupying a slot for the entire duration of the inner DAG, its own concurrency settings, and — the notorious part — a deadlock where the SubDAG task held the only pool slot while the tasks inside it waited for one. They were deprecated in 2.0 and removed in 3.0.

A `@task_group` decorator gives the TaskFlow equivalent, and groups can be nested and even generated dynamically, which makes 'one group per source system' a natural way to lay out a large DAG.

The deeper question the interviewer is often heading for is whether the DAG should be fifty tasks at all. If two halves have different schedules, different owners or different failure semantics, they are two DAGs connected by a dataset — and that is usually a better answer than a very tidy fifty-task graph.

The reasoning

`retries` counts retries, not attempts, so the total executions are `retries + 1`. The published instance shows `try 3` for a task configured with `retries=2`, which is the arithmetic worth being certain about in an interview.

Between attempts Airflow waits `retry_delay` (default five minutes), and `retry_exponential_backoff=True` grows that delay, bounded by `max_retry_delay`. Backoff is the right default for anything hitting an external service — retrying a rate-limited API three times in fifteen seconds is not a retry strategy.

The downstream task gets `upstream_failed`. It is not a failure of that task; it is Airflow recording that it can never run. Keeping the two apart matters in an incident because the `failed` tasks are the ones with logs worth reading.

The judgment half: retries are for *transient* failures. A `ConnectionError` from a vendor deserves three attempts with backoff; a `ValueError` from bad data will fail identically three times and just delays the alert by fifteen minutes. Retrying a non-idempotent task is worse than not retrying it — a partial write followed by a rerun is how you get duplicates, which is why the retry conversation always ends up back at idempotence.

The reasoning

The problem is that a partial insert leaves rows behind. The retry inserts them again, and now the day is counted one-and-a-bit times — a number nobody will notice is wrong because it is plausible.

The delete-then-insert form is the portable answer: `DELETE FROM fact_orders WHERE ds = '{{ ds }}'` followed by the insert, in one transaction so a failure between them cannot leave the table empty. The task now owns exactly one partition, and running it any number of times converges on the same content. On warehouses with `INSERT OVERWRITE` or partition replacement, use that — it is the same idea with one statement.

Where a key exists rather than a partition, `MERGE` on the key is the equivalent. And for file writes, deriving the path from the logical date — `s3://bucket/orders/ds={{ ds }}/part.parquet` — makes a rerun overwrite the same object rather than adding another one, which is the file-system version of owning your partition.

The generalisation worth stating: a task should be a function from its logical date to its output, with no dependence on what is already there. Anything that appends, increments, or reads `now()` breaks that. The check is to run it twice for the same date and diff the target — if the results differ, the retry that will inevitably happen at 3am is going to corrupt your data.

The reasoning

The mechanism is straightforward: loop over the list and assign each `DAG` object into the module globals, or build it in a function and assign the result. Airflow's parser picks up any `DAG` instance it finds at module scope, so a loop of twenty produces twenty DAGs from one file.

The danger is where the list comes from. A query to your application database at module level executes on every parse of that file — every thirty seconds, forever — and if the database is slow or unreachable the parse times out and *every* DAG in the file vanishes from the UI. That is the worst version of this failure: not an error, an intermittent disappearance.

So the list has to be cheap and local. A YAML or JSON file in the repo is the usual answer: it is version-controlled, so adding a client is a reviewable pull request, and reading it is a file read rather than a network call. Where the list genuinely must come from a system of record, generate the files in CI and commit the output, so the expensive lookup happens once per deploy rather than twice a minute.

The alternative worth weighing: often you do not want N DAGs at all. One DAG with a task mapped over the client list gives you a single schedule, one place to look, and per-client task instances with their own retries — and dynamic task mapping resolves the list at runtime rather than at parse time, which sidesteps the whole problem. Separate DAGs are right when clients need different schedules, different owners, or genuine isolation of failure.

The reasoning

`partial()` binds the arguments that do not change; only what you pass to `expand()` is mapped over. Without it you would have to build a list of identical bucket names the same length as the filenames, which is noise.

Passing two arguments to `expand()` gives you the *cartesian product*, not the pairing — `expand(a=[1,2], b=[3,4])` produces four instances, not two. That surprises people, and it is occasionally exactly what you want (every region × every day) and usually not.

When arguments vary together, `expand_kwargs()` takes a list of dictionaries, one per instance: `load.expand_kwargs([{'name': 'a.csv', 'schema': 'raw'}, ...])`. That is the shape you get naturally from an upstream task returning a list of records, which is the common real case.

Two operational notes. Mapped instances share a pool and count individually against concurrency, so `max_active_tis_per_dag` on the mapped task is what stops a wide fan-out starving the rest of the DAG. And if the upstream returns an empty list the mapped task is `skipped` with zero instances — which then propagates to anything downstream on `all_success`, and is a genuinely confusing way for a pipeline to do nothing.

The reasoning

In `poke` mode the sensor is an ordinary running task that sleeps between checks. It occupies a worker slot and a pool slot the entire time. Twenty sensors waiting six hours on a twenty-slot cluster is a cluster that does nothing else all day, and the symptom — everything queued, nothing running, no errors — is the classic Airflow deadlock.

`reschedule` mode makes the sensor exit after each check and be rescheduled by the scheduler after `poke_interval`. The slot is free in between. The cost is scheduler churn and a minimum granularity of the poke interval, so it suits long waits with infrequent checks and is wrong for a sub-minute poll.

Deferrable operators are the better answer for long waits. The task suspends into a `triggerer` process, which runs thousands of lightweight async waits in one process, and is resumed when the condition fires. No worker slot, no scheduler churn, and second-level responsiveness — `FileSensor`, `S3KeySensor` and most others have `deferrable=True` variants or async equivalents. The cost is that you must run a triggerer component.

The design question that should come first: does this need a sensor at all? Waiting six hours for another team's file is a polling solution to a notification problem. A dataset/asset trigger, an event from the producer, or the producing DAG calling yours removes the wait entirely. Sensors are for genuinely external systems you cannot get an event out of — and even then, always set `timeout`, because a sensor with no timeout waits forever.

The reasoning

The distinction first: a hook wraps *how you talk to a system* — connection resolution, authentication, retries, the client object. An operator wraps *a unit of work in a DAG* — it has a `execute(context)` and participates in templating. Most provider packages ship both, and operators are usually thin wrappers over hooks.

For twelve DAGs calling one internal API, the hook is clearly worth it: one place that resolves the connection, handles auth and normalises errors, and everything else uses it. Whether you also need an operator is a different question, and with TaskFlow the answer is often no — a `@task` function that instantiates the hook and calls it is shorter, testable as a plain function, and does not need anyone to learn a new operator's arguments.

The case for a real operator: you want fields templated with the run context (`endpoint="/reports/{{ ds }}"` only works if `endpoint` is in `template_fields`), you want a consistent interface across teams that do not write Python, or you want to encapsulate `on_kill` behaviour so a cancelled task cleans up properly. Those are real reasons; 'it feels more Airflow-y' is not.

Where to put it matters more than people think. A custom operator in your DAGs folder gets imported on every parse; putting shared code in an installed package — or at least outside the DAGs directory — keeps parsing fast and makes the code testable and versionable in CI like any other library.

The reasoning

Running heavy compute in a worker breaks the worker model. Workers are sized uniformly and shared; one task taking 64 GB either needs every worker sized for it — expensive and mostly idle — or gets OOM-killed and takes its neighbours with it. It also makes the failure opaque: a worker dying looks like a zombie task rather than like a job that ran out of memory.

The pattern is submit-and-watch. `KubernetesPodOperator` runs the work in its own pod with its own image and resource requests. `SparkSubmitOperator` or a provider's `EmrAddSteps`/`DatabricksSubmitRun` hands it to a cluster. A warehouse operator pushes the computation into the warehouse, which is the cheapest option when the work is SQL-shaped. In every case the Airflow task is small: submit, poll, report.

Make the waiting cheap too. A task that blocks a worker for two hours waiting for a Spark job has moved the problem rather than solved it — use the deferrable variant so the wait sits in the triggerer, and keep the poll interval sane.

The idempotency angle to volunteer, because it is where this goes wrong in practice: if Airflow submits a job and then dies, the retry submits it again while the first is still running. Deriving a deterministic job name or run id from the dag id and logical date lets the external system reject or reuse the duplicate — otherwise a retry doubles your cluster bill and, worse, your output.

The reasoning

`params={'start': Param('2026-03-01', type='string', format='date')}` gives you three things: a default so scheduled runs work unchanged, JSON-schema validation so a bad value is rejected before anything runs, and a generated form in the 'Trigger DAG w/ config' dialog so the analyst gets date pickers rather than raw JSON.

`dag_run.conf` is the underlying free-form dict — whatever JSON was passed at trigger time, available as `{{ dag_run.conf }}` or from the context. It is what the API and CLI send. Params are built on top of it with declaration and validation, which is why params are the answer for anything a human triggers.

The design point worth making: a scheduled run has no conf, so any code reading it must have a sensible default — `{{ params.start }}` handles that automatically, whereas `dag_run.conf['start']` raises a KeyError on every scheduled run. That asymmetry is the most common bug in parameterised DAGs.

The alternative to weigh: if the range is always 'the interval this run covers', do not parameterise at all — use `data_interval_start`/`data_interval_end` and let a backfill supply the range. Parameters are for genuinely ad-hoc runs; using them where the schedule already encodes the answer means two ways to say the same thing, and a manual run that quietly disagrees with a scheduled one.

The reasoning

The callbacks: `on_failure_callback` fires when a task instance reaches `failed`, which is after the final retry rather than after each one — so the naive worry about being paged three times is unfounded, provided you use the right hook. `on_retry_callback` is the one that fires per attempt, and it belongs in logs or a metric, not in a pager. There is also `on_success_callback`, and `on_failure_callback` at DAG level for the run as a whole.

Setting it once in `default_args` applies it to every task in the DAG, which is what you want — per-task callbacks are how half your tasks end up silent. Scoping to production is a target check inside the callback, or a shared `default_args` factory that only attaches the callback outside dev.

The callback receives the context, so the message should carry what an on-call person needs: dag id, task id, logical date, try number, and a link to the log. A Slack alert that just says 'a task failed' generates a five-minute hunt at 3am, which is exactly when you have the least patience for one.

The design caveat worth adding: a callback runs in the worker and can itself fail, and a failing callback can mask the original failure. Keep it defensive and cheap. And for anything genuinely important, do not rely on a callback at all — alert on the outcome the business cares about (the table is not fresh by 9am) rather than on the mechanism, because a DAG that never ran at all fires no task callbacks.

The reasoning

The ones that make operating possible: `owner` and `tags`, so an unfamiliar failing DAG can be routed to a human in seconds; a `description`; and a failure callback in `default_args` so nothing is silently red.

The ones that prevent incidents: `catchup=False` unless you deliberately want history, because the default has produced a lot of surprise backfills; `max_active_runs=1` for anything writing to a shared target, so a catchup or a slow day cannot have two runs fighting; and `execution_timeout` on every task that touches a network, because the alternative is a worker held forever by a hung connection.

The ones that make failures survivable: `retries` of two or three with `retry_exponential_backoff=True` for anything external, and `retry_delay` long enough to be worth waiting for. `retries=0` is the default and it means a transient blip is an incident.

The bad default I would call out beyond those is `depends_on_past` being easy to set and hard to unset — not a default exactly, but a setting that should require justification. And `start_date` deserves a rule of its own: static, timezone-aware, and never computed.

The reasoning

A DAG with a syntax error, a bad import or a cycle does not fail loudly — it fails to parse, and a DAG that does not parse simply is not there. In the UI it vanishes or goes stale; the scheduler creates no runs; nothing is red. The first person to notice is usually whoever needed the data.

The test: construct a `DagBag`, assert `import_errors == {}`, and assert `len(dagbag.dags) == expected` if you want to catch silent disappearance. Then layer on the conventions — every DAG has an owner and tags, no duplicate dag ids, every task has retries, no DAG has `catchup=True` unless it is on an allowlist. Each of those is one loop and it turns a code-review rule into something that cannot be forgotten.

It is also fast — no scheduler, no database, seconds in CI — which is what makes it the first test rather than an aspiration. Anything that imports all your DAG files also incidentally catches slow top-level code, since a test that takes a minute to collect is telling you the parser takes a minute too.

What it does not catch is anything about behaviour: the DAG parses and the logic can still be wrong. That is what unit tests on the extracted functions and an occasional `airflow dags test` against fixtures are for. But those are worth nothing if the file does not load, so this one goes in first.

The reasoning

Reasons to keep it one DAG: a single place to see the whole pipeline, dependencies enforced by the graph rather than by timing, one clear/rerun for the whole thing, and no cross-DAG coordination to get wrong. A four-hour DAG is not a problem in itself.

Reasons to split. Different schedules — ingest hourly, publish daily — cannot live in one DAG without contortions. Different owners, so a failing transform does not page the ingest team. Different retry and backfill semantics: you want to replay the transform for last March without re-ingesting. And blast radius: if a publish failure should not block tomorrow's ingest, they should not be in the same run.

The boundary to use is data, not convenience. A dataset (asset) declared by the producing task and consumed by the next DAG gives you the dependency without the coupling — the consumer triggers when the data is actually ready rather than at a guessed time, and adding a second consumer needs no change to the producer.

What I would avoid: splitting on 'it feels too big'. A sixty-task DAG that is genuinely one unit of work, with one owner and one schedule, is easier to operate as one DAG than as three connected by machinery. And splitting by layer — one DAG per team's technology rather than per data product — usually produces the worst version, where every pipeline spans three DAGs and no one DAG means anything on its own.