Chapter 12 · Operating It

Running, Failing & Re-running

Lab:Ubuntu 24.04 VM on AWS+Docker · docker-compose+SQL Server 2022 on Linux (container)+Airflow 2.7 (containers)Runs are triggered and cleared with the Airflow CLI through docker exec into the scheduler container on the Linux host.

A pipeline earns trust by how it behaves when things go wrong. This chapter replays the six runs of the rebuilt SBEcoLTDDW lab, plus a genuine bug from the lab's very first build, and shows how each was recovered without reloading anything twice.

Succeeded Failed / retried Source change (simulator)
01 initial load 9,567 orders first try ✔ 30 s first build only: ✖ update_statistics 02 no change 0 orders 41 s src_01 move + 3 new orders 03 customer moved +1 version, +3 39 s src_02 zero-value order 04 bad order ✖ dq_gate DQ04 watermark kept 37 s src_03 fix the order (49.99) 05 order fixed 1 updated 44 s 06 full reload 31,469 read 0 changed 44 s every run also deploys 11 files compresses CCI logs to audit All on 2026-10-20 between 09:58 and 10:08 UTC, after the lab was rebuilt as SBEcoLTDDW. Durations are audit.BatchRun StartedAt → EndedAt.
Figure 12.1 — Six pipeline runs and three source changes on the rebuilt Linux + Docker lab.

Triggering and watching runs

Everything here can be done in the Airflow UI (Trigger DAG w/ config, Clear). For repeatable lab runs, three tiny scripts wrap the Airflow CLI inside the scheduler container:

#!/bin/bash
# Usage: ~/sbecoltddw/airflow/sbdw_run.sh <dag_id> [conf-json] [run_id]
#   e.g. sbdw_run.sh sbecoltddw_e2e_pipeline '{"full_reload": false}' sbdw_run_01
# Task logs: ~/airflow/logs/dag_id=<dag>/run_id=<run_id>/task_id=<task>/attempt=<n>.log
set -euo pipefail
DAG="$1"; CONF="${2:-}"; RUN_ID="${3:-manual_$(date +%Y%m%dT%H%M%S)}"
[ -n "$CONF" ] || CONF='{}'
S=airflow_airflow-scheduler_1
af() { docker exec "$S" airflow "$@" 2>/dev/null; }

af dags unpause "$DAG" >/dev/null
af dags trigger "$DAG" -r "$RUN_ID" -c "$CONF" >/dev/null
echo "triggered $DAG run_id=$RUN_ID conf=$CONF"

start=$(date +%s)
while true; do
  state=$(af dags list-runs -d "$DAG" -o plain | awk -v r="$RUN_ID" '$2==r {print $3}')
  case "$state" in success|failed) break ;; esac
  if [ $(( $(date +%s) - start )) -gt 1500 ]; then echo "timeout waiting for $RUN_ID"; break; fi
  sleep 10
done
echo "run $RUN_ID finished: $state after $(( $(date +%s) - start ))s"
af tasks states-for-dag-run "$DAG" "$RUN_ID" -o plain | awk '{printf "%-48s %-16s %s %s\n", $3, $4, $5, $6}'
#!/bin/bash
# its SQL output to ~/sbecoltddw/captures/<run_id>.txt
# Usage: ~/sbecoltddw/airflow/sbdw_scenarios.sh <step>
#   move    : source change "customer moves + 3 new orders", then pipeline run 03
#   bad     : source change "zero-value order",              then pipeline run 04 (DQ gate fails)
#   fix     : source change "fix the order",                  then pipeline run 05 (passes)
#   reload  : pipeline run 06 with full_reload = true (idempotency check)
set -uo pipefail
cd "$(dirname "$0")"
step="$1"

src() {   # $1 scenario, $2 run_id
  ./sbdw_run.sh sbecoltddw_source_changes "{\"scenario\": \"$1\"}" "$2" | grep -v '^task_id'
  ./sbdw_capture.sh sbecoltddw_source_changes "$2"
}
pipe() {  # $1 run_id, $2 conf
  ./sbdw_run.sh sbecoltddw_e2e_pipeline "${2:-}" "$1" | grep -v -E 'success\s+[0-9]|skipped|^task_id'
  ./sbdw_capture.sh sbecoltddw_e2e_pipeline "$1"
}

case "$step" in
  move)   src move_and_order src_01_move_and_order;   pipe sbdw_run_03_customer_moved ;;
  bad)    src bad_order      src_02_bad_order;        pipe sbdw_run_04_bad_order ;;
  fix)    src fix_bad_order  src_03_fix_bad_order;    pipe sbdw_run_05_order_fixed ;;
  reload) pipe sbdw_run_06_full_reload '{"full_reload": true}' ;;
  *) echo "unknown step $step"; exit 2 ;;
esac

The first build: a real bug, recovered in place

When this lab was first built, its very first run failed, and the failure was not staged. It was fixed the way you would in production: fix the SQL file, then clear the failed task and its downstream tasks inside the same DAG run. Upstream work (deploy, extracts, dimensions, the fact load) was not repeated. (After the fix, the rebuilt SBEcoLTDDW lab's run 01 succeeds on the first try.)

  1. Attempt 1 of post_load.update_statistics failed. Incorrect syntax near the keyword 'statistics': the file used AS statistics as a column alias, and STATISTICS is a reserved word in T-SQL. Everything before it (all the loads) had succeeded; everything after it became upstream_failed and fail_run marked the batch Failed.
  2. Fix the file. The alias became stats_objects, and the file was copied to ~/airflow/sql on the Linux host. Because the bind mount makes it visible to the containers immediately, and SQL is read at task run time, no restart was needed.
  3. Clear and re-run. Clearing from update_statistics with --downstream re-ran only the last 7 tasks. The run ended success and end_run set the batch to Succeeded.
#!/bin/bash
# Usage: ~/sbecoltddw/airflow/sbdw_clear.sh <dag_id> <run_id> <task_regex>
set -euo pipefail
DAG="$1"; RUN_ID="$2"; TASK="$3"
S=airflow_airflow-scheduler_1
af() { docker exec "$S" airflow "$@" 2>/dev/null; }

LD=$(af dags list-runs -d "$DAG" -o plain | awk -v r="$RUN_ID" '$2==r {print $4}')
echo "clearing $TASK (+downstream) in $RUN_ID (logical date $LD)"
af tasks clear "$DAG" -t "$TASK" --downstream --yes -s "$LD" -e "$LD" | tail -3

start=$(date +%s); sleep 5
while true; do
  state=$(af dags list-runs -d "$DAG" -o plain | awk -v r="$RUN_ID" '$2==r {print $3}')
  case "$state" in success|failed) break ;; esac
  if [ $(( $(date +%s) - start )) -gt 1500 ]; then echo "timeout"; break; fi
  sleep 10
done
echo "run $RUN_ID finished: $state after $(( $(date +%s) - start ))s"
af tasks states-for-dag-run "$DAG" "$RUN_ID" -o plain | awk '{printf "%-48s %-16s\n", $3, $4}'

Why clearing was safe

Every task is idempotent. When the statistics task was cleared, the loads that had succeeded were not re-run, and the tasks that ran afterwards would have produced the same result even if they had already run once. The batch's Attempts stayed 1 because start_run wasn't cleared. Clear from start_run if you want the retry counted.

Runs 02 and 03: nothing, then one move

Run 02 had nothing to do and did nothing: 0 changed orders, new customers 0, new versions 0. Then the simulator moved customer 11090 and placed 3 orders:

move_customer      before | 5781 Sharon Dr. | Los Angeles | 90012 | 2012-01-07 00:00:00
                   SQL> customer 11090 moved to Seattle WA
                   after  | 600 4th Ave     | Seattle     | 98104 | 2026-10-20 10:00:58.940
insert_new_orders  SQL> 3 new orders for customer 11090
                   75124 | SO75124 | 2026-10-20 10:01:01.350 | 752.3400
                   75125 | SO75125 | 2026-10-20 10:01:01.350 | 753.3400
                   75126 | SO75126 | 2026-10-20 10:01:01.350 | 754.3400

Run 03 extracted exactly those 3 orders, expired one customer version and created one (Chapter 6), and inserted the 3 orders on the new version (Chapter 7).

Run 04: the gate says no

The simulator inserted a zero-value order. Run 04 loaded it, the gate's rule DQ04 failed, and the failure path took over:

run sbdw_run_04_bad_order finished: failed after 65s
dq_gate                     failed
publish.advance_watermarks  upstream_failed
publish.run_summary         upstream_failed
end_run                     upstream_failed
fail_run                    success
BatchID | Status | StartedAt               | EndedAt
4       | Failed | 2026-10-20 10:03:39.708 | 2026-10-20 10:04:16.959

SourceObject                              | watermark_kept          | not_committed
AdventureWorks2022.Sales.SalesOrderHeader | 2026-10-20 10:01:01.350 | 2026-10-20 10:02:49.947

Run 05: fix at the source, run again

The right fix for bad source data is in the source. The simulator's fix_bad_order set the total to 49.99 and moved ModifiedDate forward. Run 05 started from the kept watermark, re-extracted the order, and the MERGE updated the existing fact row (inserted 0, updated 1). The gate passed and the watermark moved to 10:04:37.053.

Don't "fix" a gate failure by editing the warehouse

Updating Fact.Orders by hand would make the gate pass once, then the next full reload would bring the bad value back from the source. Fix the source (or add a documented correction rule to the pipeline), then re-run.

Run 06: the idempotency proof

Triggered with {"full_reload": true}, the extract ignored the watermark and re-read all 31,469 orders. Result: inserted 0, updated 0, unchanged 9571, no new customer versions, and no new dates. Being able to reload everything and change nothing is the best evidence that the incremental logic is right.

All runs, from the audit table

BatchID | DagRunID                  | FullReload | Status    | Attempts | StartedAt               | EndedAt                 | secs
1       | sbdw_run_01_initial_load   | 0          | Succeeded | 1        | 2026-10-20 09:58:36.781 | 2026-10-20 09:59:06.938 | 30
2       | sbdw_run_02_no_changes     | 0          | Succeeded | 1        | 2026-10-20 09:59:50.828 | 2026-10-20 10:00:31.043 | 41
3       | sbdw_run_03_customer_moved | 0          | Succeeded | 1        | 2026-10-20 10:01:48.125 | 2026-10-20 10:02:27.633 | 39
4       | sbdw_run_04_bad_order      | 0          | Failed    | 1        | 2026-10-20 10:03:39.708 | 2026-10-20 10:04:16.959 | 37
5       | sbdw_run_05_order_fixed    | 0          | Succeeded | 1        | 2026-10-20 10:05:28.572 | 2026-10-20 10:06:12.008 | 44
6       | sbdw_run_06_full_reload    | 1          | Succeeded | 1        | 2026-10-20 10:07:00.902 | 2026-10-20 10:07:44.509 | 44

About 40 seconds per run, and almost all of it is Airflow scheduling 26 tasks one wave after another (1–2 s between tasks). The SQL itself is a few seconds in total. That overhead is fixed per task, so the same DAG on ten times the data would take only a little longer.

Going to production

  • Schedule it: set schedule="0 2 * * *", keep catchup=False and max_active_runs=1. The watermark design means a skipped night is simply caught up by the next run.
  • Alert on failure: add on_failure_callback (Slack, Teams) or email_on_failure in default_args. fail_run already records the failure in SQL for dashboards.
  • Retries: retries=2, retry_delay=timedelta(minutes=5) on extract tasks absorbs transient network or lock errors. The gate should keep retries=0, because bad data won't fix itself.
  • SLAs and freshness: alert when etl.Watermark.LastValue falls too far behind the source's newest ModifiedDate.

Key takeaways

  • Fix the SQL file, then clear the failed task with its downstream; upstream work is kept.
  • A failed gate leaves the watermark in place, so the next run re-reads the same changes.
  • Fix bad data at the source and re-run; never patch the warehouse by hand.
  • A full reload that changes nothing proves the pipeline is idempotent.