Orchestration Patterns
Task granularity, dependencies that skip rather than fail, retries with backoff, and the sensor pattern that waits for data instead of hoping it arrived.
Orchestration is the layer that decides what runs, in what order, and what happens when something fails. The patterns below are tool-agnostic — they apply equally in Airflow, Dagster, Prefect or a well-written shell script.
Task granularity
# one task, doing everything
def nightly(run_date):
ingest_orders(run_date)
ingest_customers(run_date)
build_silver(run_date)
build_marts(run_date)
export_to_bi(run_date)[2026-09-09 02:00:04] ingest_orders ok (42s)
[2026-09-09 02:00:46] ingest_customers ok (18s)
[2026-09-09 02:01:04] build_silver ok (204s)
[2026-09-09 02:04:28] build_marts ok (188s)
[2026-09-09 02:07:36] export_to_bi FAILED: connection refused
Task nightly FAILED after 456sThe export failed and the whole task is red. Retrying re-runs seven and a half minutes of successful work to reach the one step that needs it — and if the ingest is not idempotent, the retry corrupts the data.
Split at the boundaries you would want to retry:
TASKS = {
"ingest_orders": (ingest_orders, []),
"ingest_customers": (ingest_customers, []),
"build_silver": (build_silver, ["ingest_orders", "ingest_customers"]),
"build_marts": (build_marts, ["build_silver"]),
"export_to_bi": (export_to_bi, ["build_marts"]),
}ingest_orders SUCCESS 42s
ingest_customers SUCCESS 18s
build_silver SUCCESS 204s
build_marts SUCCESS 188s
export_to_bi FAILED 12s connection refusedNow the retry runs twelve seconds of work. The two ingests also ran concurrently, because neither depends on the other.
A minimal runner
The whole model in forty lines, which is worth writing once to understand what a scheduler is doing:
# runner.py
import time
from collections import defaultdict
def run_dag(tasks, run_date, max_retries=2):
state = {name: "pending" for name in tasks}
def ready(name):
return all(state[dep] == "success" for _, deps in [tasks[name]] for dep in deps)
def blocked(name):
return any(state[dep] in ("failed", "skipped") for _, deps in [tasks[name]] for dep in deps)
while "pending" in state.values():
progressed = False
for name, (fn, deps) in tasks.items():
if state[name] != "pending":
continue
if blocked(name):
state[name] = "skipped"
print(f"{name:<18} SKIPPED (upstream failed)")
progressed = True
elif ready(name):
for attempt in range(1, max_retries + 2):
try:
t0 = time.perf_counter()
fn(run_date)
state[name] = "success"
print(f"{name:<18} SUCCESS {time.perf_counter()-t0:5.1f}s")
break
except Exception as e:
if attempt > max_retries:
state[name] = "failed"
print(f"{name:<18} FAILED {e}")
else:
wait = 2 ** attempt
print(f"{name:<18} retry {attempt} in {wait}s ({e})")
time.sleep(wait)
progressed = True
if not progressed:
raise RuntimeError(f"deadlock: {state}")
return state
print(run_dag(TASKS, "2026-01-04"))ingest_orders SUCCESS 0.4s
ingest_customers SUCCESS 0.2s
build_silver SUCCESS 2.0s
build_marts SUCCESS 1.9s
export_to_bi retry 1 in 2s (connection refused)
export_to_bi retry 2 in 4s (connection refused)
export_to_bi FAILED connection refused
{'ingest_orders': 'success', ..., 'export_to_bi': 'failed'}Three behaviours worth naming: dependencies gate execution, failures skip dependents rather than failing them, and retries back off exponentially.
Skip, do not cascade
def build_silver(run_date):
raise RuntimeError("source schema changed")
run_dag(TASKS, "2026-01-04")ingest_orders SUCCESS 0.4s
ingest_customers SUCCESS 0.2s
build_silver retry 1 in 2s (source schema changed)
build_silver retry 2 in 4s (source schema changed)
build_silver FAILED source schema changed
build_marts SKIPPED (upstream failed)
export_to_bi SKIPPED (upstream failed)One red task naming the real problem, two grey ones. The alternative — three failures, three alerts, three stack traces — buries the cause and pages three times for one incident.
The data consequence matters as much as the alerting one: build_marts did not run, so the
marts still hold yesterday’s correct numbers rather than being rebuilt from a broken silver.
That is the same SKIP behaviour dbt build and a Snowflake task DAG give you.
Retries only on idempotent tasks
def ingest_orders_bad(run_date):
con.execute("insert into orders select * from read_csv(?)", [f"landing/{run_date}.csv"])
if random.random() < 0.5:
raise RuntimeError("network blip after the write")ingest_orders retry 1 in 2s (network blip after the write)
ingest_orders SUCCESS 0.3sGreen, and the data is doubled — the first attempt wrote before it failed. Retries amplify non-idempotence. Configure them only where the task is safe to repeat, which in practice means: fix idempotency first, then enable retries.
Retry the transient, not the deterministic:
TRANSIENT = (ConnectionError, TimeoutError)
def should_retry(exc):
return isinstance(exc, TRANSIENT)build_silver FAILED BinderException: Column "amount" not foundNo retries burned on a schema error that will fail identically three more times — the alert arrives immediately instead of six seconds later.
Sensors
The worst dependency is a guess:
# "the loader usually finishes by 01:30, so run at 02:00"
schedule = "0 2 * * *"One slow day and the pipeline processes a partial file, passes every check, and publishes low numbers. Wait for the data instead:
def wait_for_partition(run_date, timeout_s=3600, poll_s=60):
deadline = time.time() + timeout_s
while time.time() < deadline:
ready = con.execute("""
select count(*) from load_history
where load_date = ? and status = 'complete'
""", [run_date]).fetchone()[0]
if ready:
print(f"partition {run_date} is ready")
return
print(f"waiting for {run_date} ...")
time.sleep(poll_s)
raise TimeoutError(f"partition {run_date} not ready after {timeout_s}s")
TASKS["wait_for_orders"] = (wait_for_partition, [])
TASKS["build_silver"] = (build_silver, ["wait_for_orders", "ingest_customers"])waiting for 2026-01-04 ...
waiting for 2026-01-04 ...
partition 2026-01-04 is ready
build_silver SUCCESS 2.0sTwo rules for sensors. Always set a timeout, or a sensor waiting for data that will never arrive holds a worker forever. And prefer event-driven triggering where the platform offers it — Airflow datasets, Dagster asset sensors, a message on completion — since polling costs a worker slot per waiting sensor.
Parameterise everything
def build_marts(run_date):
con.execute("delete from daily_revenue where ordered_at = ?", [run_date])
con.execute("""insert into daily_revenue
select ordered_at, count(*), sum(amount) from silver_orders
where ordered_at = ? and status = 'completed' group by 1""", [run_date])Every task takes the window as an argument and nothing reads the clock. That single discipline is what makes the next section possible.
Backfills are the same DAG
from concurrent.futures import ThreadPoolExecutor
from datetime import date, timedelta
days = [date(2026, 1, 1) + timedelta(days=i) for i in range(7)]
with ThreadPoolExecutor(max_workers=3) as pool:
results = list(pool.map(lambda d: (d, run_dag(TASKS, d.isoformat())), days))
for d, state in results:
failed = [k for k, v in state.items() if v == "failed"]
print(f"{d} {'FAILED: ' + ', '.join(failed) if failed else 'ok'}")2026-01-01 ok
2026-01-02 ok
2026-01-03 FAILED: build_silver
2026-01-04 ok
2026-01-05 ok
2026-01-06 ok
2026-01-07 okOne day failed and six succeeded. Re-running just the 3rd completes the backfill, because each day is independent and idempotent.
max_workers=3 is not decoration. An unbounded backfill against a production source is an
outage — and the person running it is usually already dealing with an incident.
Alert on the right things
con.execute("""
create table if not exists task_runs (
dag varchar, task varchar, run_date date, status varchar,
duration_s double, attempts int, finished_at timestamp
)
""")
print(con.execute("""
select task,
count(*) filter (status = 'failed') as failures,
round(avg(duration_s), 1) as avg_s,
round(avg(duration_s) filter (run_date >= current_date - 7), 1) as avg_s_7d,
max(attempts) as max_attempts
from task_runs group by 1 order by failures desc, avg_s desc
""").df().to_string(index=False)) task failures avg_s avg_s_7d max_attempts
export_to_bi 3 12.4 14.1 3
build_marts 0 188.2 301.7 1
build_silver 0 204.1 206.0 1Two findings a green pipeline hides. export_to_bi fails regularly and is rescued by retries —
flaky, not fine. And build_marts has gone from 188s to 302s over a week, which is the shape
of a job about to breach its window.
Alert on: a task failing after retries, a run breaching its SLA, a success with zero rows, and a duration well above its trailing average. Do not alert on every retry.
In a real orchestrator
# Airflow, for comparison — the same DAG
with DAG("bookshop_nightly", schedule="0 2 * * *", catchup=True,
default_args={"retries": 2, "retry_delay": timedelta(minutes=5),
"retry_exponential_backoff": True}) as dag:
wait = SqlSensor(task_id="wait_for_orders", timeout=3600, poke_interval=60, sql="...")
silver = PythonOperator(task_id="build_silver", python_callable=build_silver,
op_kwargs={"run_date": "{{ ds }}"})
marts = PythonOperator(task_id="build_marts", python_callable=build_marts,
op_kwargs={"run_date": "{{ ds }}"})
wait >> silver >> marts{{ ds }} is the logical date from lesson 4, catchup=True turns the same DAG into a
backfill, and the skip-on-failure behaviour is the default. The site’s airflow track covers
this in depth; the patterns above are what you are configuring.
Practice
1. Split a monolithic task and fail the last step.
build_marts SUCCESS 188s
export_to_bi FAILED 12sThe retry costs twelve seconds instead of seven minutes. Task boundaries are retry boundaries — that is the whole reason to draw them.
2. Fail a middle task and watch dependents skip.
build_silver FAILED
build_marts SKIPPED
export_to_bi SKIPPEDOne alert instead of three, and the marts keep yesterday’s correct data. A DAG that cascades failures instead of skipping trains people to ignore alerts.
3. Add retries to a non-idempotent task.
ingest_orders retry 1 in 2s (network blip after the write)
ingest_orders SUCCESSGreen, and the rows are duplicated. Idempotency is a prerequisite for retries, not an optimisation to add later.
4. Replace a scheduled delay with a sensor.
waiting for 2026-01-04 ...
partition 2026-01-04 is readyThe pipeline now starts when the data exists rather than when someone guessed it would. Set a timeout — a sensor with none is a worker leak with a plausible explanation.
Next: streaming and exactly-once — the same problems with the clock running.