Data pipelines and orchestration (Apache Airflow basics)

Monday 9 am: Sunday’s revenue is ₹0

What are data pipelines and orchestration?

Words you’ll meet in this lesson

An Airflow DAG (example)

This is what the daily orders pipeline looks like in Airflow 3 using the TaskFlow style. It is shown for reading; it needs an Airflow installation to run.

from datetime import datetime, timedelta
from airflow.sdk import dag, task      # Airflow 2.x: from airflow.decorators import dag, task

@dag(schedule="0 1 * * *",                         # every day at 1:00 am
     start_date=datetime(2026, 10, 1),
     catchup=False,
     default_args={"retries": 3, "retry_delay": timedelta(minutes=5)})
def daily_orders():

    @task
    def extract(ds=None):                           # ds = the logical date, e.g. "2026-10-19"
        return f"s3://lake/bronze/orders/date={ds}/"

    @task
    def validate(path):
        ...                                         # row counts, nulls, negative amounts
        return path

    @task
    def transform(path):
        ...                                         # submit a Spark job: bronze -> silver -> gold

    transform(validate(extract()))

daily_orders()

Pipelines and orchestration at three levels

Kinds of pipelines, and cron vs an orchestrator

Cron vs an orchestrator

Building a reliable daily pipeline

1. Draw the DAG

   dag = {"extract": [], "validate": ["extract"], "transform": ["validate"],
          "publish": ["transform"], "email_report": ["publish"]}
   done, order = set(), []
   while len(order) < len(dag):
       for task, deps in dag.items():
           if task not in done and all(d in done for d in deps):
               order.append(task); done.add(task)
   print(" -> ".join(order))

   extract -> validate -> transform -> publish -> email_report
   Each task lists what it depends on; the orchestrator works out a valid order (a topological sort) and runs independent tasks in parallel when it can.

2. Parameterise by logical date

   import datetime
   logical = datetime.date(2026, 10, 19)
   path = f"s3://lake/bronze/orders/date={logical}/"
   print("this run reads and writes only:", path)

   this run reads and writes only: s3://lake/bronze/orders/date=2026-10-19/
   The 19 October run reads and writes only 19 October’s partition, whether it runs at 1 am on the 20th or during a backfill next month.

3. Extract with retries

   The source API sometimes times out at night. The extract task is set to retries=3 with a 5-minute delay (often growing each time: exponential backoff).

   Most night-time failures are temporary: a busy API, a network blip, a restarting database. Retries turn them into short delays instead of morning emergencies.

4. Validate before publishing

   rows = [120, 340, -90, 410]
   checks = {"has rows": len(rows) > 0, "no negative amounts": all(r >= 0 for r in rows)}
   print(checks, "-> PUBLISH" if all(checks.values()) else "-> STOP and alert")

   {'has rows': True, 'no negative amounts': False} -> STOP and alert
   A quality gate between transform and publish stops impossible data (a negative amount, zero rows, a sudden 90% drop) and alerts someone. Publishing nothing is better than publishing wrong numbers.

5. Publish idempotently

   partitions = {}
   def publish(day, revenue):
       partitions[day] = revenue          # overwrite the day’s partition
   publish("2026-10-19", 850); publish("2026-10-19", 850)   # re-run
   print(partitions, "total:", sum(partitions.values()))

   {'2026-10-19': 850} total: 850
   Overwriting the date partition means a retry or a re-run produces the same result. With appends, every re-run would add the day’s revenue again.

6. Monitor, alert, backfill

   Airflow records every run. If Sunday failed, an alert goes out, the dashboard shows “data delayed”, and once fixed, the team clears and re-runs Sunday: a one-day backfill.

   Good pipelines are judged on the bad days: how fast failures are noticed, how clearly they are reported and how safely they are repaired.

One night: what the managers see, and what Airflow does

Step through the six frames. Left: the dashboard the managers will open. Right: the DAG run behind it, including a failure and a retry.

Real-time scenarios: pipelines on bad days

Three live scenarios: a failure that retries can’t fix, a week-long backfill after a bug, and an advanced look at logical dates and late-arriving data.

A failure that retries can’t fix

Retries fix most night-time hiccups. But on Diwali night the source database is overloaded for hours. Watch how a well-built pipeline fails safely and recovers.

Backfilling a week after a bug fix

A bug under-counted wallet payments for seven days. The code is fixed; now the past must be repaired. Watch why idempotent tasks make a backfill safe, and appends make it a disaster.

Logical dates, late data and sensors

Real data doesn’t arrive on time. Orders from phones that were offline can show up days later. Watch how logical dates, a reprocessing window and a sensor keep daily numbers correct.

Where orchestration runs every day

A pipeline as a DAG, run in order

The pipeline is a dictionary of tasks and the tasks each depends on. The program finds a valid order (a topological sort), runs each task for one logical date, retries a flaky step, and stops downstream tasks if something fails for good. Then try the challenges: parallel branches, quality gates, idempotent publishing, backfills and incremental loads.

dag = {"extract": [], "validate": ["extract"], "transform": ["validate"], "publish": ["transform"]}
attempts_needed = {"extract": 2}            # extract fails once, then works
RETRIES = 3

def order(dag):
    done, out = set(), []
    while len(out) < len(dag):
        for t, deps in dag.items():
            if t not in done and all(d in done for d in deps):
                out.append(t); done.add(t)
    return out

logical_date = "2026-10-19"
for task in order(dag):
    for attempt in range(1, RETRIES + 1):
        if attempt >= attempts_needed.get(task, 1):
            print(f"{logical_date} {task:9} success (attempt {attempt})")
            break
        print(f"{logical_date} {task:9} failed, retrying")
    else:
        print(f"{task} failed for good: downstream skipped, alert sent")
        break

• dag = {"extract": [], "validate": ["extract"], "transform": ["validate"], "publish": ["transform"]}Each task lists its upstream tasks. That is all an orchestrator needs to know the order.
• if t not in done and all(d in done for d in deps):A task is ready only when every task it depends on has finished.
• for attempt in range(1, RETRIES + 1):Retries handle temporary failures such as timeouts without waking anyone up.
• else:The for … else runs only if no attempt succeeded: stop the pipeline instead of running on missing data.

Run it on your PC: a tiny orchestrator with retries and backfills

This program defines a four-task DAG, runs it in dependency order for a logical date, retries a flaky extract step, and publishes one JSON file per day into a warehouse folder. Then it backfills three days twice to prove that re-running is safe. Only the standard library is used. Save it as mini_orchestrator.py and run it.

"""A tiny orchestrator: tasks with dependencies, run in order, with retries,
one output partition per logical date, and a backfill that re-runs safely."""
import json, os, random

random.seed(1)
os.makedirs("warehouse", exist_ok=True)

ORDERS = {  # pretend source system: orders per day
    "2026-10-17": [120, 340, 90], "2026-10-18": [200, 150], "2026-10-19": [410, 80, 60, 300],
}
DAG = {"extract": [], "validate": ["extract"], "transform": ["validate"], "publish": ["transform"]}


def topo_order(dag):
    done, order = set(), []
    while len(order) < len(dag):
        for task, deps in dag.items():
            if task not in done and all(d in done for d in deps):
                order.append(task)
                done.add(task)
    return order


def run_task(task, day, state, flaky):
    if task == "extract":
        if flaky and random.random() < 0.6:      # the source API sometimes times out
            raise TimeoutError("source API timed out")
        state["rows"] = ORDERS[day]
    elif task == "validate":
        assert all(a > 0 for a in state["rows"]), "negative amount"
    elif task == "transform":
        state["summary"] = {"day": day, "orders": len(state["rows"]), "revenue": sum(state["rows"])}
    elif task == "publish":
        # idempotent: overwrite this day's partition instead of appending
        with open(f"warehouse/daily_{day}.json", "w") as f:
            json.dump(state["summary"], f)


def run_dag(day, flaky=False, retries=3, verbose=True):
    state = {}
    for task in topo_order(DAG):
        for attempt in range(1, retries + 1):
            try:
                run_task(task, day, state, flaky)
                if verbose:
                    print(f"  {day} {task:9} ok" + (f" (attempt {attempt})" if attempt > 1 else ""))
                break
            except Exception as e:
                print(f"  {day} {task:9} failed: {e}" + ("; retrying" if attempt < retries else "; giving up"))
        else:
            print(f"  {day} downstream tasks skipped, alert sent")
            return
    if not verbose:
        print(f"  {day} all tasks ok -> {state['summary']}")


print("Task order:", topo_order(DAG))
print("\nNightly run for 2026-10-19 (flaky source):")
run_dag("2026-10-19", flaky=True)

print("\nBackfill 2026-10-17 to 2026-10-19 (run twice to prove it is safe):")
for run in (1, 2):
    print(f" run {run}:")
    for day in ORDERS:
        run_dag(day, verbose=False)
files = sorted(os.listdir("warehouse"))
total = sum(json.load(open("warehouse/" + f))["revenue"] for f in files)
print(f"\nPartitions in warehouse: {files}")
print(f"Total revenue across partitions: Rs {total} (not doubled, because publish overwrites)")

# In a terminal:  python mini_orchestrator.py

Finding the run order, one line at a time

Three tasks are listed out of order. Watch the loop pick a task only when everything it depends on is already done.

Worked examples: real problems, solved and traced in memory

Four pipeline problems, from ordering tasks to sizing a backfill. Each runs live and is traced.

0 1 * * * · 30 */4 * * * · 0 9 * * 1

0 1 * * *     -> minute 0, hour 1, every day
30 */4 * * *  -> minute 30, every 4 hours, every day
0 9 * * 1     -> minute 0, hour 9, on Mondays

extract 10, clean_orders 20, clean_riders 5, report 8 (minutes)

wave: ['extract']
wave: ['clean_orders', 'clean_riders']
wave: ['report']
shortest total: 38 minutes (vs 43 one by one)

4 retries, backoff 2 min doubling, 3 min per attempt

delays between attempts: [2, 4, 8, 16]
worst case before failing: 45 minutes
fits a 2-hour window

45 days, 12 min per run, 6 parallel, 6 hours

batches: 8 | minutes needed: 96
nights required: 1

Check yourself

Quick revision

Interview and viva questions

Practice problems

Pipeline and orchestration exercises. The last two are business tasks.

DAG of 4 tasks

['clean', 'enrich']

dag = {"extract": [], "clean": ["extract"], "enrich": ["extract"], "report": ["clean", "enrich"]}
print(dag["report"])

2026-10-19

s3://lake/gold/revenue/date=2026-10-19/

ds = "2026-10-19"
print(f"s3://lake/gold/revenue/date={ds}/")

every 4 hours

2190

print(24 // 4 * 365)

DAG of 5 tasks

['enrich', 'report']

dag = {"extract": [], "clean": ["extract"], "enrich": ["clean"], "report": ["enrich"], "audit": ["extract"]}
failed = {"clean"}
changed = True
while changed:
    changed = False
    for t, deps in dag.items():
        if t not in failed and any(d in failed for d in deps):
            failed.add(t); changed = True
print(sorted(failed - {"clean"}))

ids [40, 41, 43, 44, 47]

load: [43, 44, 47] | new watermark: 47

ids = [40, 41, 43, 44, 47]
last = 42
new = [i for i in ids if i > last]
print("load:", new, "| new watermark:", max(new))

a → b → c → a

cycle!

graph = {"a": ["b"], "b": ["c"], "c": ["a"]}
def has_cycle(g):
    visiting, done = set(), set()
    def visit(n):
        if n in visiting:
            return True
        if n in done:
            return False
        visiting.add(n)
        found = any(visit(m) for m in g[n])
        visiting.discard(n); done.add(n)
        return found
    return any(visit(n) for n in g)
print("cycle!" if has_cycle(graph) else "no cycle")

2026-10-19, window 3

['2026-10-19', '2026-10-18', '2026-10-17']

import datetime
run = datetime.date(2026, 10, 19)
print([str(run - datetime.timedelta(days=d)) for d in range(3)])

25, 40, 70, 15 min

normal: done at 3:30 OK
worst: done at 4:40 OK

normal = 25 + 40 + 70 + 15
worst = normal + 2 * (25 + 10)
for name, m in [("normal", normal), ("worst", worst)]:
    end = 60 + m          # minutes after midnight
    print(f"{name}: done at {end // 60}:{end % 60:02d}", "OK" if end <= 8 * 60 else "LATE")

5 × 3 h × Rs 1,500 × 2 per month

cost per month: 45000
payback months: 2.7

monthly_cost = 5 * 3 * 1500 * 2
print("cost per month:", monthly_cost)
print("payback months:", round(1_20_000 / monthly_cost, 1))

Real-world scenario questions (asked in company interviews)

Pipeline and orchestration questions from real teams and interviews. Decide your answer first, then open the model answer.