Orchestration — Airflow, DAGs, Retries, Backfills

Why this session exists

Orchestration coordinates dependencies, scheduling, retries and historical runs. It does not make arbitrary writes atomic or correct. This lesson targets Airflow 2.10.5 and preserves the full daily-order pipeline, sensor, retry, backfill, data-quality and alerting exercises while making their boundaries explicit. Airflow 3 imports, APIs and alerting changes are not silently assumed compatible.

You will define DAG/task/operator/XCom, choose stable interval inputs, implement retry-safe publication, plan a historical backfill, configure bounded retry/timeout/SLA behavior, and compare Airflow, Prefect, Dagster and Argo against deployment and modeling needs.

Prerequisites

Batch versus streaming explains finite-run input contracts; Spark supplies an example external compute engine. Airflow can coordinate finite tasks around continuously produced data; it is not an event-at-a-time processor.

A coordinator, not an atomicity engine

A wedding coordinator tracks prerequisites and completion, while caterers do the work. Airflow similarly records tasks and dependencies, schedules attempts, and observes outcomes. A DAG is a directed acyclic workflow definition with scheduling/configuration, not necessarily one file: a Python module may define multiple DAGs. A task is a unit in that definition; a task instance is its concrete execution context within a run, with attempts. An operator defines how work is performed; TaskFlow decorators wrap Python functions into tasks. XCom communicates small values such as a URI, count or manifest identity between task instances.

Airflow workers can legitimately perform bounded Python work; “never compute inside a worker” is too absolute. Submit large Spark/warehouse/container work to an appropriate engine and budget worker resources. A submission must identify and wait for the external job, and recover a lost completion acknowledgment without submitting duplicate effects.

DAG files are parsed repeatedly according to parsing/scheduler settings, not necessarily every scheduler heartbeat or exactly every 30 seconds. Keep top-level network calls, database queries and expensive computation out of DAG import. Heavy imports can also delay parsing. Constructing a DAG object at import is expected; executing its tasks then is not [Airflow best practices].

Historical and version context

Airflow began at Airbnb and became an Apache project. TaskFlow and dynamic mapping are not inventions of a supposed 2024 Airflow 3 release. This lesson deliberately uses the 2.10.5 API family, its SLA behavior and its dags backfill CLI. When migrating versions, inspect the matching release documentation rather than substituting “stable” links into an older example.

Anatomy of an Airflow deployment

The executor decides where queued work runs. Celery uses a broker/worker arrangement; KubernetesExecutor schedules task pods and should not be mislabeled as the same Celery queue. The metadata database tracks orchestration state, not the bulk business datasets. Deployment may separate DAG processors and schedulers; neither the diagram nor a TaskFlow function defines a universal process topology.

A tiny DAG walkthrough

The original illustrative branching DAG—extract, stage, independent customer/order transforms, validate both, publish, notify—teaches a fan-in: publication waits for all required branches. Our executable reference uses a simpler order-only chain with the same quality-before-publication invariant. Arrows specify dependencies; they do not carry data or turn separate external effects into one transaction. Default all_success and alternative trigger rules change downstream scheduling, so inspect them before calling validation a gate.

Retries, backfills, SLAs and sensors

Retries: retries=3 permits three retries after the initial attempt: up to four attempts. With constant five-minute retry delays, waiting between attempts totals 15 minutes, plus attempt durations, not an automatic 20 minutes. Exponential backoff and max_retry_delay bound retry spacing; pools and concurrency limits also protect a struggling dependency. on_retry_callback is the callback name, not retry_callback. Permanent invalid input should fail clearly rather than be retried indefinitely; use an intentional fail-without-retry policy where appropriate.

Backfills: catchup creates scheduled runs for eligible missed data intervals; an explicit backfill selects historical logical dates even when regular catchup is disabled. It need not rerun already successful tasks without the relevant reset/rerun choices. catchup=False does not forbid manual historical execution. max_active_runs, task concurrency, pools and the DAG's max_active_runs limit bound pressure; catchup does not imply 730 simultaneous runs.

SLA versus timeout: Airflow 2.10.5 task documentation describes completion relative to a DAG-run start; the tagged implementation traverses scheduled timetable intervals and compares next_info.logical_date + task.sla with current time. This is not an attempt-runtime stopwatch. Check the deployed timetable and observed SLA records rather than deriving a wall-clock deadline from the task's start time. SLA misses are notifications/records and do not kill tasks; execution timeout limits an attempt's running time, and sensor timeout has its own waiting behavior. Scheduled-run SLA monitoring is not a complete data-freshness check. Configure recipients/callbacks/integrations explicitly; an SLA declaration does not magically send Slack or PagerDuty [Airflow tasks].

Sensors: poke mode occupies a worker slot while checking; reschedule releases it between checks. Supported deferrable operators hand waiting to a triggerer and release the worker, with executor/pool semantics to verify. An ExternalTaskSensor normally matches the same logical date, not “whatever upstream ran most recently.” Different upstream schedules may require execution_delta or an explicit date mapping. Choose allowed/failed/skipped states and timeout deliberately.

Data interval, logical date and idempotence

For an @daily UTC schedule, the run created after 2024-08-11 00:00 normally covers [2024-08-10 00:00,2024-08-11 00:00) and has logical date at interval start. A cron 0 3 * * * instead defines 03:00-to-03:00 intervals under its timetable; extracting only logical_date.date() does not magically mean the previous calendar day. Use explicit timezone-aware interval boundaries and align the upstream contract [Airflow runs].

Same logical date and code alone do not imply same output: input snapshot, reference data, randomness and configuration must also be stable. Define retry identity from pipeline, interval and input/code version. A retry repeats the same intended publication; a correction/backfill may intentionally publish a new version. A now() used for logging is fine; a now() used to choose business input breaks historical reproducibility.

Derive retry safety from ambiguous completion: a worker can commit an effect and disappear before Airflow records success. Retrying may repeat the effect; refusing to retry can leave work incomplete. Rather than assuming exactly one execution, design a recoverable operation—atomic partition replacement, deterministic versioned publication, or a dedup identity recorded with the effect. A plain INSERT can be safe under a unique immutable-event contract; an UPSERT can be unsafe if it increments values. Names of SQL verbs are not proofs of idempotence.

If four files are published separately and attempt one writes three, a retry with four new random names leaves partial/duplicate output. Stage under a stable versioned identity, validate the full set, then atomically publish a manifest/pointer using the storage system's supported primitive. Dependencies alone cannot provide that atomicity.

XCom and task boundaries

Use XCom for a manifest URI, input version, row count or external-job ID. There is no universal safe 10KB limit independent of backend/configuration; large DataFrames are still the wrong default because of serialization/storage/scheduler pressure. Object-storage XCom backends change mechanics, not the need for ownership, retention and access controls.

Fine tasks isolate failures and expose branches but add scheduling and durable intermediate-output costs. Coarse tasks let one engine optimize across transformations but give coarse orchestration visibility and may repeat more work. Engine-level checkpoints can preserve recovery granularity inside a coarse task. Split at useful failure/publication boundaries, not every Python line or necessarily every network call; one business transaction may correctly include several database statements.

Hands-on: exact retry-safe publication and quality gate

This complete Python 3/SQLite program models the business behavior of the DAG with immutable fixture inputs and atomic day replacement. It covers unchanged retries, corrected input including deletions/empty output, failure before commit, and a seven-day volume gate. The same pure check_volume definition is included in the reference DAG below for its optional volume-gate extension; that extension still needs the governed seven-day history wired in. The function itself is executed exactly, not a second paraphrased implementation.

import sqlite3
from fractions import Fraction
 
def check_volume(current, history):
    if len(history)!=7 or any(x<0 for x in history) or current<0:
        raise ValueError('seven nonnegative daily counts required')
    mean=Fraction(sum(history),7)
    if mean==0:
        if current!=0: raise ValueError('new activity needs explicit approval')
    elif not Fraction(4,5)*mean <= current <= Fraction(6,5)*mean:
        raise ValueError('outside inclusive 20 percent band')
    return True
 
c=sqlite3.connect(':memory:')
c.execute('CREATE TABLE mart(day TEXT,order_id TEXT,amount_cents INT NOT NULL CHECK(amount_cents>=0),PRIMARY KEY(day,order_id))')
def publish(day, rows, fail=False):
    if len({oid for oid,n in rows})!=len(rows): raise ValueError('duplicate order identity')
    if any(n<0 for oid,n in rows): raise ValueError('invalid amount')
    with c:
        c.execute('DELETE FROM mart WHERE day=?',(day,))
        c.executemany('INSERT INTO mart VALUES(?,?,?)',[(day,oid,n) for oid,n in rows])
        if fail: raise RuntimeError('before commit')
 
day='2024-08-10'; original=[('O1',1500),('O2',600)]
publish(day,original);publish(day,original)
assert c.execute('SELECT sum(amount_cents) FROM mart').fetchone()[0]==2100
try: publish(day,[('O1',999)],True)
except RuntimeError: pass
assert c.execute('SELECT sum(amount_cents) FROM mart').fetchone()[0]==2100
# A correction removes O2. MERGE-upserts alone would leave that stale row.
publish(day,[('O1',1400)])
assert c.execute('SELECT order_id,amount_cents FROM mart').fetchall()==[('O1',1400)]
publish(day,[])
assert c.execute('SELECT count(*) FROM mart').fetchone()[0]==0
for n in [80,100,120]: assert check_volume(n,[100]*7)
for n in [79,121]:
    try: check_volume(n,[100]*7)
    except ValueError: pass
    else: raise AssertionError('volume gate failed')
assert check_volume(0,[0]*7)
try: check_volume(1,[0]*7)
except ValueError: pass
else: raise AssertionError('zero-baseline policy failed')
# Validation before publish protects the last accepted snapshot.
publish(day,original)
try: publish(day,[('O1',-1)])
except ValueError: pass
assert c.execute('SELECT sum(amount_cents) FROM mart').fetchone()[0]==2100
assert 3*5==15
print({'retry_total_cents':2100,'correction_deletes_stale_order':True,'empty_partition_supported':True,'volume_boundary':[80,120],'retry_wait_minutes':15})
c.close()

Airflow 2.10.5 reference DAG

This is a source-reviewed integration reference, not an executed scheduler/database deployment. It uses an operator-managed libpq service orders_lesson through psycopg2, avoiding unpinned provider-hook APIs and embedded credentials. Require matching UTC daily upstream intervals and precreated tables: lake_manifest(snapshot_id,order_day,is_complete) unique by snapshot/day, plus lake_orders(snapshot_id,order_day,order_id,amount_cents) immutable by snapshot; stage_orders(day,order_id,amount_cents) and mart_orders(day,order_id,amount_cents) each unique on (day,order_id). params.snapshot_id identifies the selected immutable fixture/version. Concurrent external writers are excluded; production needs per-partition serialization or an atomic version pointer.

from datetime import timedelta
from contextlib import closing
from fractions import Fraction
import pendulum
import psycopg2
from airflow.decorators import dag, task
from airflow.operators.python import get_current_context
from airflow.sensors.external_task import ExternalTaskSensor
 
# Same pure function as the executed local lab.
def check_volume(current, history):
    if len(history)!=7 or any(x<0 for x in history) or current<0:
        raise ValueError('seven nonnegative daily counts required')
    mean=Fraction(sum(history),7)
    if mean==0:
        if current!=0: raise ValueError('new activity needs explicit approval')
    elif not Fraction(4,5)*mean <= current <= Fraction(6,5)*mean:
        raise ValueError('outside inclusive 20 percent band')
    return True
 
@dag(dag_id='orders_pipeline',schedule='@daily',
 start_date=pendulum.datetime(2024,1,1,tz='UTC'),catchup=False,max_active_runs=1,
 params={'snapshot_id':'fixture-v1'},
 default_args={'retries':3,'retry_delay':timedelta(minutes=5),
 'retry_exponential_backoff':True,'max_retry_delay':timedelta(minutes=30),
 'execution_timeout':timedelta(minutes=30)})
def orders_pipeline():
    wait=ExternalTaskSensor(task_id='wait_ingest',external_dag_id='ingest_orders',
      external_task_id='publish_to_lake',allowed_states=['success'],
      failed_states=['failed','upstream_failed'],skipped_states=['skipped'],
      mode='reschedule',poke_interval=60,timeout=4*60*60)
 
    @task
    def stage():
        ctx=get_current_context(); day=ctx['data_interval_start'].date()
        snapshot=ctx['params']['snapshot_id']
        with closing(psycopg2.connect(service='orders_lesson')) as conn:
            with conn,conn.cursor() as cur:
                cur.execute('SELECT is_complete FROM lake_manifest WHERE snapshot_id=%s AND order_day=%s',(snapshot,day))
                if cur.fetchone() != (True,):
                    raise ValueError('missing or incomplete input snapshot')
                cur.execute('DELETE FROM stage_orders WHERE day=%s',(day,))
                cur.execute('''INSERT INTO stage_orders(day,order_id,amount_cents)
                  SELECT order_day,order_id,amount_cents FROM lake_orders
                  WHERE order_day=%s AND snapshot_id=%s''',(day,snapshot))
                cur.execute('SELECT count(*) FROM stage_orders WHERE day=%s',(day,))
                count=cur.fetchone()[0]
        return {'day':day.isoformat(),'count':count,'snapshot':snapshot}
 
    @task(retries=0)
    def validate(info):
        with closing(psycopg2.connect(service='orders_lesson')) as conn:
            with conn,conn.cursor() as cur:
                cur.execute('SELECT count(*) FROM stage_orders WHERE day=%s AND (amount_cents IS NULL OR amount_cents<0)',(info['day'],))
                if cur.fetchone()[0]: raise ValueError('invalid staged amount')
        # Optional volume gate must receive seven finalized day counts,
        # including valid zeros, from a governed completion manifest.
        # Call check_volume(info['count'], history) after loading that manifest.
        return info
 
    @task
    def publish(info):
        with closing(psycopg2.connect(service='orders_lesson')) as conn:
            with conn,conn.cursor() as cur:
                cur.execute('DELETE FROM mart_orders WHERE day=%s',(info['day'],))
                cur.execute('''INSERT INTO mart_orders(day,order_id,amount_cents)
                  SELECT day,order_id,amount_cents FROM stage_orders WHERE day=%s''',(info['day'],))
        return info
 
    staged=stage(); checked=validate(staged); published=publish(checked)
    wait >> staged
orders_pipeline()

What each block does and the exercise answers

The timezone-aware daily schedule avoids confusing a 03:00 interval with a calendar-day partition. Sensor matching assumes the upstream's same UTC logical-date schedule; if its interval differs, a tested mapping is required. reschedule releases workers between checks; skipped upstream work skips the sensor under the explicit policy. TaskFlow function arguments create dependencies; wait >> staged adds the external prerequisite. The diagram is an explanatory workflow, not a screenshot claim.

Stage first requires a completed input manifest, distinguishing legitimate empty input from missing/unfinished input, then replaces exactly one input-version/day partition. Parameter binding protects values and removes SQL string interpolation. Validation happens before publication, so bad new data cannot overwrite the last accepted mart merely because a later DQ task fails. Publish deletes and inserts one partition atomically, handling removed orders and legitimate empty corrections. The old MERGE read the entire stage, not just its intended day, and did not delete removed orders; AirflowSkipException on zero rows would leave stale facts. Skipping is appropriate only when the business contract truly says “no publication,” not as a substitute for correct empty output.

The reference excludes concurrent writers: max_active_runs=1 constrains regular DAG runs but is not a database lock against manual jobs, another DAG or backfill/executor configurations. Immutable input, stable stage identity and partition serialization are production requirements. A connection transaction provides database atomicity, not a cross-system transaction with Airflow metadata.

Volume-gate extension: load the seven preceding finalized day counts including zeros from completion manifests, then call the exact check_volume. With seven counts of 100, 80 and 120 pass, 79 and 121 fail. With seven zeros, this policy accepts zero and escalates positive activity for explicit review. Missing history is not the same as zero; incomplete history should follow a separate approved cold-start rule. Weekday/holiday seasonality may justify a different metric. On failed validation, an all_success publication is blocked as upstream-failed; skipped and upstream-failed are different states. Alerts happen only if configured.

Backfill extension: for an authorized historical fixture, Airflow 2.10.5 uses:

airflow dags backfill orders_pipeline -s 2024-08-01 -e 2024-08-07

Review existing task states and the CLI's reset/rerun behavior before invoking it. Select the input snapshot appropriate to each interval; the sample single fixture snapshot is not a complete historical manifest service. Repeating a successful interval against identical bytes yields identical output; intentionally corrected input publishes a new accepted result. Test failure before database commit and after commit/before scheduler acknowledgment, not just a happy-path run.

Production reality and alerting

Repeated parsing scenario: top-level API calls are multiplied by actual parse activity and can burden both scheduler and remote service. Move them into tasks and measure parse duration; do not attribute an unsupported exact 2,880-calls/day incident to Airbnb.

Backfill scenario: appending the same data on retry duplicates it; a unique-key error may instead stop the pipeline. Atomic partition replacement or a recoverable versioned publication gives a concrete retry contract. A claimed four-day Astronomer customer recovery is not needed to establish that mechanism.

Catchup scenario: a historical start date with catchup enabled can enqueue many eligible runs, but execution is bounded by configured concurrency/pools. Choose catchup intentionally; false/one are reasonable fixture defaults, not universally mandatory. Plan warehouse capacity and cancellation/recovery of queued work rather than assuming every historical run starts at once.

Finish-by-06:00 strategy: for the UTC daily schedule, the run first eligible after August 11 midnight represents August 10. The required publication deadline is August 11 at 06:00, regardless of when a worker starts. Record that deadline alongside the expected interval and committed output manifest. An independent monitor checks the expected manifest/version and completion timestamp at the deadline, including legitimate zero-row days; it must not equate row_count=0 with failure.

For an advisory Airflow SLA, set the task/operator's sla=timedelta(...) (or the inherited default_args) and the DAG's sla_miss_callback, with tested recipients/integration. Read the tagged manage_slas traversal and verify the resulting SLA-miss execution date for your timetable and successful-run history. Its scheduled-interval scan and the high-level “Dag Run start time” wording do not license either the original “1.5 hours from 3am” shortcut or an untested arithmetic replacement. The reference DAG therefore does not pretend an advisory SLA is the authoritative 06:00 gate. Use execution_timeout for a runaway attempt, sensor timeout for the wait, and the independent deadline check for missing/failed/stale/partial publications. Test all four conditions and notification delivery in the actual deployment; no alert was sent here.

Choosing an orchestrator

Compare deployment constraints, workflow/data model, and integration/operations:

• Airflow 2.10.5: explicit task DAGs, scheduled data intervals/backfills and broad operator ecosystem; maintain parsing, metadata, executor and dependency contracts.
• Prefect 3: Python flows/tasks with tracked runs, retries and deployment infrastructure; useful when Python control flow and operational run management fit the team.
• Dagster: asset definitions/dependencies and materialization-oriented modeling; evaluate partitions, lineage and testability for the data estate.
• Argo Workflows 3.6: Kubernetes-native container steps/DAGs; good when Kubernetes is the intentional execution platform, with cluster/RBAC/artifact operations to own.

No universal hiring/testability ranking selects a winner. A simple cron job can be correct if its dependencies/retries/publication are explicitly managed; an Airflow DAG can be incorrect if they are not.

Where this connects next

Airflow can run dbt transformations and DQ gates, reconcile CDC outputs, submit Kubernetes work, monitor delivery deadlines, or orchestrate model training/evaluation. Those tools still own their internal semantics. Observability should measure accepted data freshness and external job state, not only green task boxes.

Recall and worked answers

1. DAG/task/operator? Workflow definition; unit and its run-specific instance; implementation of how work runs. XCom carries small coordination values.
2. Why retry-safe? Completion can be ambiguous and historical reruns are deliberate; one intended effect must survive repeat attempts.
3. Concrete write pattern? Bound parameters for a stable interval, validate immutable input, then DELETE+INSERT that interval in one transaction—or a versioned/dedup protocol with equivalent semantics.
4. Deploy settings? Deliberate start date/timezone/timetable, catchup policy and concurrency/pools. False/one are fixture choices, not proof against every concurrent writer.
5. XCom for what? Pointers and small metadata; no universal byte threshold makes arbitrary payloads a good design.
6. Poke versus reschedule? Occupy the worker while waiting versus release between checks; deferral uses a triggerer for supported operators.
7. Three retries at five minutes danger? Permanent failures waste up to 15 minutes of delay plus execution time; distinguish transient from poison input and bound attempts/timeouts.
8. Airflow versus Dagster? Task/run-oriented orchestration versus an asset-centric model is a useful starting distinction; evaluate deployment, partitions/integrations and team operations, not an automatic winner.
9. Deadline alerting? Derive SLA against the actual logical-date/timetable behavior and independently validate the intended interval's committed output at 06:00; configure and test notifications, including valid empty data.

File-upload failure answer: dependencies did not guarantee atomic publication. Stage a deterministic version, verify all four files, then publish one manifest/pointer through a supported atomic operation. A retry must not invent another visible output set.

Sources and next step

Checked 2026-10-07: Airflow 2.10.5 best practices, DAG runs, tasks and SLAs, DAGs, CLI, Prefect 3 flows, Dagster assets, Argo 3.6. Tagged Airflow source supplies precise SLA/sensor checks in the evidence.

Next: dbt warehouse ELT. Exact local transaction/quality models ran; Airflow scheduling, provider sends, PostgreSQL and external job execution were not run. Source-supported orchestration teaching is complete within that boundary.