analytics now has all its computing pieces: the lake in HDFS fed by upload_events_hdfs.py (04-02), daily_sales.py in Spark (05-03), the recommendations with ALS, and the Flink streams that need nobody to launch them because they never finish (05-04). What is missing is whatever makes the batch jobs happen every day without anyone launching them by hand: waiting for the day's file to be complete, checking that it is not corrupt, running the spark-submit, loading the result into the database the dashboards query, raising the alarm if something fails, retrying without duplicating, and doing it all again for seven days when a price is corrected. The initial version of that at Kilometre Zero was a four-line crontab and a run_all.sh script, and it failed in every possible way: the file arrived late and Spark processed half a day, a retry loaded the sales twice, and nobody noticed until a producer asked why their chart had doubled. This lesson treats the data pipeline as what it is, one more distributed system: a DAG of tasks with dependencies, time-based and data-based scheduling, retries, idempotency, backfill, SLAs and data quality; it introduces Apache Airflow as the reference scheduler (and mentions Prefect, Dagster and Argo Workflows); and it builds dags/daily_sales.py, the daily pipeline of analytics, which closes the module.

1. From loose scripts to pipelines
2. The concepts: DAG of tasks, scheduling, retries, idempotency, backfill, SLAs
3. cron and its limits
4. Apache Airflow: architecture and programming model
5. Alternatives: Prefect, Dagster and Argo Workflows
6. Data quality, lineage and catalogue
7. Hands-on: dags/daily_sales.py
8. Backfilling Grape Harvest Week
9. Common Mistakes and Tips
10. Exercises
11. Conclusion

The original crontab of analytics was this:

Each line is correct, and the whole is fragile for reasons we already know from the rest of the course:

• The dependencies are implicit and time-based. Spark starts at 3:00 assuming that the 2:00 upload has finished. The day the upload takes 70 minutes (a campaign spike), Spark processes a half-written file and the 4:00 load publishes incomplete sales. It is the zero-latency fallacy (01-04) applied to the duration of a job.
• There are no retries, or there are retries without idempotency. If load_postgres.py fails halfway, somebody relaunches it by hand and the rows are duplicated, because it does an INSERT with no key.
• Reprocessing is manual. Correcting seven days means editing seven dates by hand in three commands, in order, without making a mistake.
• There is no observability. cron sends an email with the standard output if the process returns a non-zero exit code, and nothing else. How long did it take yesterday? Did it run on the 12th? Which task fails most often?
• The state lives in one machine's crontab. If that machine dies (fallacy 1), there is no pipeline; if two people edit it, nobody knows which version is running.

A data pipeline solves this by making the implicit explicit: the tasks and their dependencies form a graph, each task declares when it can run (by time, or when the data it needs exists), what to do if it fails and how long it may take, and the system that runs it records every execution and lets you relaunch any stretch of it. It is task parallelism (05-01, section 2) with a scheduler that knows the complete DAG.

The DAG of tasks. Tasks are nodes and dependencies are directed edges; acyclic because a task cannot depend on itself. Unlike Spark's DAG of operators (05-03), here each node is a complete job (a spark-submit, a database load, an HTTP call) and the scheduler does not move data between nodes: the nodes communicate through storage (HDFS, PostgreSQL) and only exchange small pieces of metadata (how many rows, which path). Tasks with no mutual dependency run in parallel; the total time is set by the critical path.

Time-based and data-based scheduling. "At 3:00" is time-based scheduling, and it is not enough when the input is produced by another system. Data-based (or event-based) scheduling adds sensors: tasks that do nothing except wait for a condition to become true (the file /km0/events/2026-09-14/orders.jsonl exists; there is a row in a table; another DAG has finished; a message has arrived on a queue) and that fail if the condition is not met within a deadline. The DAG starts at 3:00 but Spark does not begin until the sensor confirms that the file is there.

Retries with backoff. Many failures are transient (a NameNode in failover, a saturated PostgreSQL, a timeout). Each task declares how many times to retry and with what wait, a growing one (exponential backoff with a cap and jitter, the same policy 02-05 gave to consumers). Deterministic failures (a bug in the code) exhaust the retries and then do fail.

Idempotency of every task. Retries and reprocessing are only safe if running a task twice for the same day leaves the same result as running it once. The universal technique is writing by partition with overwrite: the task for day D produces exactly partition D of its output (the day=2026-09-14/ directory in Parquet, the rows with day = '2026-09-14' in PostgreSQL) and replaces it entirely, inside a transaction or with an atomic rename. Never an append, never an INSERT without deleting first or without ON CONFLICT. daily_sales.py already does this with partitionOverwriteMode=dynamic (05-03); the load into PostgreSQL will do it with DELETE ... WHERE day = %s followed by an INSERT in the same transaction.

Backfill. Running the pipeline for a range of past dates: because an input was corrected (the Grape Harvest Week prices), because the logic changed (a new column that has to be filled in historically), or because the pipeline was down. It is only possible if every run is parameterised by its logical date (not by "yesterday") and if the tasks are idempotent. A pipeline that uses date -d yesterday cannot be backfilled.

SLAs and alerts. An SLA (service level agreement) for the pipeline is "the sales for day D are in PostgreSQL before 6:00 on day D+1". The scheduler watches the deadline and warns when a task or the DAG misses it, as well as warning on every definitive failure. Alerts go to the team's channel; failures, to whoever is on call (07-01 will deal with monitoring in general).

Versioning the pipeline. The DAG is code in the repository (km0/dags/), reviewed and deployed like everything else: you know which version ran each day, and a change of logic is a commit, not an edit to the crontab. Ideally, the code version is recorded alongside every run.

cron is still the right tool for "run this command at this time on this machine" when there are no dependencies and no state to manage: rotating logs, a backup, a reminder. As a pipeline orchestrator, the comparison is this:

Airflow (Airbnb, 2014; Apache since 2016) defines pipelines as Python code and runs them with this architecture:

• Scheduler. The heart. It periodically parses the files in dags/, works out for each DAG which runs (DAG runs) are due according to its schedule and its start_date, and for each run which tasks have their dependencies met; it queues them on the executor. More than one can be run for high availability.
• Executor. How tasks get run: LocalExecutor (processes on the scheduler's machine: enough for development and modest pipelines), CeleryExecutor (a queue, Redis or the RabbitMQ of 02-04, and workers on several machines), KubernetesExecutor (one pod per task, 07-05).
• Workers. They run each task's code. For tasks that launch work on another system (Spark, a query), the worker just waits and supervises; the heavy computation does not happen in Airflow.
• Metadata database. PostgreSQL holding the state of everything: which DAG runs exist, what state each task instance is in, how many attempts, the XComs. It is the source of truth, and that is why Airflow does not lose the pipeline if a worker dies.
• Webserver. The UI: the graph, the grid of runs per day, logs per attempt, and the buttons for relaunching, marking as success or clearing.

The programming model:

• DAG. A DAG object with an id, a schedule (a cron expression, a timedelta, or a dataset for data-based scheduling), a start_date, catchup and default_args for the tasks.
• Operators and tasks. Each task is an instance of an operator: BashOperator, PythonOperator, SparkSubmitOperator, SQLExecuteQueryOperator, hundreds more in the providers. Sensors are operators that wait: FileSensor, WebHdfsSensor, ExternalTaskSensor, SqlSensor. Dependencies are declared with >>.
• Logical date and templates. Each DAG run has a logical date (logical_date, formerly execution_date) and a data interval (data_interval_start/end). With a daily schedule, the run that processes the data for 14 September has the logical date 2026-09-14 and runs when the interval ends, that is, on the 15th at the scheduled time. The {{ ds }} macro in any templated field evaluates to 2026-09-14 for that run: this is what makes backfill possible, because a run for the 8th will have ds = 2026-09-08 even if it is launched in October.
• catchup. If it is True, when a DAG with a start_date in the past is enabled, the scheduler creates and runs one run for every interval not executed since then. It is useful for populating history and dangerous if you are not expecting it (hundreds of runs at once). With False, it only runs from the current interval onwards, and the history is done with an explicit backfill.
• depends_on_past. A run's task does not start until the same task in the previous run has finished successfully. Necessary when each day builds on the previous one (a running total); unnecessary and harmful when the days are independent, because a failure on the 12th blocks the 13th, the 14th...
• retries, retry_delay, retry_exponential_backoff, max_retry_delay. The per-task retry policy.
• XCom. A mechanism for one task to leave a small value (a row count, a path) and another to read it, through the metadata database. For metadata, not for data: anything larger than a few KB goes to storage and its path travels through XCom.
• Pools. Named concurrency quotas: a spark pool with 2 slots guarantees that there are never more than two spark-submit jobs at once even if twenty backfill runs ask for them.
• max_active_runs, concurrency, trigger_rule (by default all_success: the task starts if all the previous ones succeeded; all_done, one_failed... for clean-up or notification tasks).

Airflow is the de facto standard, with the weight that comes with it: a database, a scheduler, an ageing UI, and a model (DAG runs per time interval) designed for daily batches. The alternatives go after its weak points:

The choice for Kilometre Zero is Airflow for its maturity, for the Spark, HDFS and PostgreSQL operators that already exist, and because the team knows it; Dagster would be the serious alternative if the data platform grew to dozens of derived datasets. The concepts (DAG, logical date, per-partition idempotency, sensors, backfill) are the same in all four.

A pipeline that correctly runs a computation over bad data produces bad results on time. Data quality is verified inside the pipeline, as tasks:

• Input validations, before spending compute: the file exists and is closed (nobody is still writing it), it has a plausible size (a campaign day with 200 lines is suspicious), a sample parses as JSON, the mandatory fields are there, the fraction of corrupt lines is below a threshold, the event dates fall on the expected day.
• Output validations, before publishing: the total amount matches the job's check figure (05-03 printed a Check: for that reason), there are no unknown producers, there are no negative amounts, the number of rows is within the historical range (±50% of the equivalent day the week before).
• Quarantine. An input file that does not pass validation is neither half-processed nor discarded: it is moved to /km0/quarantine/<day>/ with a report saying why, the task fails with a clear message, and somebody decides. That day's valid data can be reprocessed with a backfill once the input is corrected.

Tools such as Great Expectations or Soda express those checks declaratively and integrate with Airflow; for this lesson's pipeline a Python task will do. Two more concepts that we only mention: lineage (from which inputs and with which code each output was produced: daily_sales/day=2026-09-14 comes from events/2026-09-14/orders.jsonl and from catalog.csv with version a3f9 of daily_sales.py; OpenLineage standardises it and Airflow emits it) and the data catalogue (the inventory of which datasets exist, their schema, their owner and their freshness: the Hive Metastore of 05-02, DataHub, Amundsen). Both are what lets you answer "where does this number come from?" without reading code.

Airflow is added to the docker-compose.yml of km0/ with its own metadata database and LocalExecutor (enough here; in production, Celery or Kubernetes):

After docker compose run airflow airflow db migrate and creating a user, the UI is at localhost:8090. The target table in km0_analytics, with the key that makes the load idempotent:

The points worth fixing in your mind:

• The logical date governs everything. {{ ds }} in the sensor and in the Spark arguments, ds as a parameter of the Python functions. The run on 15 September at 3:00 has ds = 2026-09-14 and processes the file for the 14th: it is the data interval that has just closed. Nothing in the DAG says "yesterday".
• The sensor in reschedule mode does not take up an executor slot while it waits: it is rescheduled every 5 minutes. With mode="poke" (the default) the worker would be blocked for six hours. The timeout turns "the file never arrived" into a visible failure at 9:00 instead of a hung pipeline.
• validate_input fails with AirflowFailException, which does not retry: a corrupt file is not fixed by waiting five minutes, and the file has already been moved to quarantine. The other failures (a network exception while reading HDFS) do retry according to default_args.
• Idempotency in all three writes. Spark overwrites only day={{ ds }} (05-03); load_postgres deletes and inserts the day in one transaction; the quarantine uses rename, atomic in HDFS. Relaunching any task, or the entire run, leaves the same state.
• XCom for metadata. lines, rows, total: numbers, not data. The data travels through HDFS and PostgreSQL.
• pool="spark" limits simultaneous Spark jobs to two even if a backfill creates seven runs at once (section 8), and max_active_runs=1 keeps normal operation to one day at a time.
• notify with trigger_rule="all_done" runs whether the run went well or some task failed, and that is why it can report the state; without that rule, a failure in validate_input would leave it in upstream_failed and nobody would find out. The SLA in default_args adds an alert if any task has still not finished three hours after 3:00.

When the DAG is enabled in the UI, the scheduler creates the first run at the next scheduled time (with catchup=False), and the grid shows one square per day and task, green, red or yellow (retrying). A click on a square gives the logs of that attempt, including the spark-submit output with the explain() of 05-03.

The case of exercise 3 in 05-03, now with the pipeline: orders corrects the price of crianza-wine and regenerates the orders.jsonl files from 8 to 14 September in HDFS. Those seven days have to be recomputed, and only those, without touching the rest and without interfering with the daily run. With the DAG parameterised by logical date and idempotent, it is one command:

What happens: the scheduler creates (or resets) seven DAG runs with ds from the 08th to the 14th and runs them respecting each one's dependencies and the global limits: the spark pool allows two compute_sales at once, so the seven Spark jobs run in four rounds; max_active_runs does not apply to the backfill (it has its own limit, --max-active-runs in recent versions, or the DAG's max_active_runs depending on the version: it is worth checking, because a one-year backfill with seven runs at once can saturate HDFS). The sensors pass immediately (the files exist), the validations are repeated over the corrected files, Spark overwrites day=2026-09-08/ ... day=2026-09-14/ and leaves the other days untouched, and each load deletes and inserts its day. When it finishes, the grid shows the seven days with a new attempt in green, and SELECT day, SUM(amount) FROM daily_sales WHERE day BETWEEN '2026-09-08' AND '2026-09-14' GROUP BY 1 reflects the corrected prices.

Two useful variants: to re-run only from the load onwards (Spark ran fine, PostgreSQL failed), airflow tasks clear km0_daily_sales -t load_postgres --downstream -s 2026-09-08 -e 2026-09-14 clears that task and the following ones in those runs, and the scheduler re-runs them; and for a one-off run with a specific date, airflow dags trigger km0_daily_sales --logical-date 2026-09-14. None of the three was possible with the crontab of section 1 without editing commands by hand.

• Using "yesterday" instead of the logical date. datetime.now() or date -d yesterday inside a task make backfill impossible and produce different results depending on when it runs. Always {{ ds }} / data_interval_start.
• Confusing when a run executes with which data it processes. The run with logical date the 14th executes on the 15th. It is the biggest source of confusion in Airflow; thinking of "the interval that has just closed" clears it up.
• Non-idempotent tasks. An INSERT without deleting first, an append to a file, a counter. Retrying duplicates; backfill accumulates. A whole partition with overwrite, always.
• Heavy computation inside the Airflow worker. A PythonOperator that reads 150 MB of JSON and aggregates in pandas turns the worker into a badly sized compute node. Airflow orchestrates; Spark, the database or a container computes.
• Data through XCom. A serialised DataFrame in the metadata database. XCom is for paths and counters.
• catchup=True by accident. Enabling a DAG with a start_date two years ago and the default catchup launches 730 runs. catchup=False and an explicit backfill.
• depends_on_past=True as a precaution. One failed day blocks all the following ones until somebody fixes it. Only when each day genuinely depends on the previous one.
• Sensors in poke mode with long timeouts. Each sensor takes up an executor slot for hours; ten waiting sensors block the entire pipeline. mode="reschedule", or dataset-based scheduling.
• Retrying deterministic errors. A corrupt file or a bug gets retried three times with backoff and fails an hour later. AirflowFailException for anything that waiting will not fix.
• No output validation. A green pipeline does not mean correct data. A check on totals and a plausible range against history cost twenty lines and prevent doubled charts.
• Secrets in the DAG. PostgreSQL passwords in the code. Airflow connections, environment variables or a secrets manager (06-04).

Extend the DAG so that, after validate_output, it trains the recommendations with recommendations_als.py (05-03) on the clicks of the last 7 days and loads the result into the table recommendations(customer, product, affinity, generated_at) in km0_catalog. Decide: which sensor does it need? Which tasks does it depend on? How do you make the load idempotent if the table has no natural "partition by day"? Should it use depends_on_past? What happens to the recommendations during a seven-day backfill?

On 20 September at 3:00 the following happens: upload_events_hdfs.py had a failure and the file /km0/events/2026-09-19/orders.jsonl reaches HDFS at 7:40 with 180,000 lines, of which 4,200 are not valid JSON. Describe, task by task, what the DAG of section 7 does between 3:00 and 9:00: states, retries, where the file ends up, what the team sees. Afterwards, somebody corrects the file and uploads it again at 11:30. Which command does the team run to complete the 19th, and what happens to the run for the 20th that night?

For each of these Kilometre Zero jobs, decide whether you would leave it in cron, put it in Airflow, or move it out to a stream as in 05-04, and justify it in one sentence: (a) deleting Flink checkpoints more than 30 days old from HDFS; (b) generating the weekly PDF report per producer from daily_sales every Monday, and sending it by email; (c) recomputing the minimum stock per product and market every 5 minutes; (d) exporting a copy of km0_inventory to MinIO every night; (e) loading into PostgreSQL every day the late positions from delivery.positions that the stream diverted to the side output.

Exercise 1.

New tasks: wait_for_clicks (a WebHdfsSensor on /km0/clicks/{{ ds }}/ or, better, on a _CLOSED closing file that the hourly upload process writes when the day is over; without it, the directory exists from the first hour and the sensor would pass with incomplete data), train_als (a SparkSubmitOperator with recommendations_als.py and an argument with the range {{ macros.ds_add(ds, -6) }} to {{ ds }}), load_recommendations (a PythonOperator). Dependencies: wait_for_clicks >> train_als >> load_recommendations, and train_als also downstream of validate_output only if the recommendations use the sales (they do not: they use clicks), so strictly speaking they are two parallel branches that converge on notify. Idempotency without a partition by day: the table is replaced entirely in one transaction (DELETE FROM recommendations; INSERT ...; COMMIT), or you write to recommendations_new and do an ALTER TABLE ... RENAME swapping the two (atomic in PostgreSQL), or you keep a column generated_at = ds and the website reads WHERE generated_at = (SELECT MAX(generated_at) ...): the third option is the one that allows backfill without trampling on the current version. depends_on_past: no; each training run is independent. During a seven-day backfill, the seven runs would train seven models with moving windows of clicks and load seven times: with the generated_at column that is harmless but pointless; the sensible thing is for train_als to live in another DAG with its own cycle (daily, with no backfill unless requested), or for the recommendations branch to be excluded from the backfill with --task-regex.

Exercise 2.

3:00: the scheduler creates the run ds=2026-09-19. wait_for_events checks every 5 minutes (reschedule, without holding a worker) and does not find the file: state up_for_reschedule, yellow square. 6:00: the 3 h SLA is missed and Airflow records an SLA miss with a warning to the team (the pipeline has not failed yet, but it is running late). 7:40: the file appears; on the 7:45 check the sensor moves to success. validate_input starts: 180,000 lines (> 1,000), but 4,200 corrupt ones are 2.3% (> 1%): it moves the file to /km0/quarantine/2026-09-19/orders.jsonl, writes report.txt and raises AirflowFailException: state failed with no retries. compute_sales, load_postgres and validate_output are left in upstream_failed. notify runs (all_done) and publishes "daily_sales 2026-09-19: FAILED at validate_input"; the failure callback also raises the alarm. At 9:00 nothing else happens: the sensor's timeout no longer applies because the sensor finished. The team sees the red square on validate_input, reads the log with "4200 corrupt lines out of 180000" and finds the file in quarantine.

11:30: the corrected file is uploaded to /km0/events/2026-09-19/orders.jsonl. The team runs airflow tasks clear km0_daily_sales -t validate_input --downstream -s 2026-09-19 -e 2026-09-19 (or clicks Clear on the task in the UI): validate_input and the following tasks go back to None and the scheduler re-runs them; the sensor is not repeated because it was not cleared. With max_active_runs=1, that run holds the slot until it finishes (a few minutes). That night, at 3:00 on the 21st, the run ds=2026-09-20 is created as normal: depends_on_past=False, so it is not affected by what happened with the 19th, and the 19th is already green.

Exercise 3.

(a) cron (or Flink itself with checkpoint retention configured): a command with no dependencies and no data, hdfs dfs -rm with a date; if a day is skipped, nothing happens. (b) Airflow: it depends on daily_sales having loaded the seven days (an ExternalTaskSensor on the daily DAG), it has output that needs to be regenerable (a one-week backfill if data is corrected) and a send that must not be duplicated. (c) Stream (05-04): every 5 minutes with seconds of latency and per-key state is a tumbling window over stock.updated, not a batch launched 288 times a day. (d) Airflow, even though it is a single task: you want history, an alert if it fails and a sensor or validation that the copy completed (size, _SUCCESS); in cron it would be acceptable if external monitoring were added. (e) Airflow: it is a daily batch parameterised by date that reads /km0/delivery/late/{{ ds }}/ and loads by partition, and it fits as one more task in the daily pipeline: it is the reconciliation between the stream and the batch that exercise 1 of 05-04 talked about.

A data pipeline is the part of the distributed system that turns loose jobs into a platform: a DAG of tasks with explicit dependencies, scheduled by time and by data (sensors), with retries with backoff for anything transient and immediate failure for anything deterministic, idempotent tasks that write whole partitions with overwrite, backfill parameterised by the logical date and not by "yesterday", SLAs and notifications to know when something is running late, input and output validations with quarantine so as not to publish bad data on time, and all of it in the repository, versioned. cron offers none of that; Airflow offers it with a scheduler, an executor, workers, a metadata database and DAGs in Python, and Prefect, Dagster and Argo Workflows recast it with different emphases. dags/daily_sales.py chains the sensor on /km0/events/{{ ds }}/orders.jsonl, the validation, the SparkSubmitOperator with the program from 05-03, the transactional load into km0_analytics and the check on totals, and backfilling Grape Harvest Week comes down to one command with two dates.

That closes Module 5, and Kilometre Zero's data platform is complete:

The data is spread out (Module 4) and is processed in bulk, in batch and in real time, without anybody launching anything by hand. But there is something we have taken for granted in every module so far: any process could talk to any service and read any data. daily_sales.py reads the whole lake; load_postgres gets into km0_analytics with a password in an environment variable; the Flink consumer reads orders.events without identifying itself; an hdfs dfs -rm from any container would wipe out a year of events; and MinIO's presigned URLs (04-03) were the only access mechanism we have designed with care. In a system with dozens of services, hundreds of jobs and the personal data of Anna, Mark and Lucy, that cannot go on. Module 6 deals with security in distributed systems, and begins with the two questions no system can avoid: who is who (authentication) and who can do what (authorization).