Tasks fail at work: a source doesn't answer, a file holds a bad value. When clean_orders fails, build_report must not run: it would build the report from orders that were never cleaned, and send_email would mail that wrong report to the team. An orchestrator stops every task that waits for a failed task, directly or through other tasks, and still runs the rest.
The tasks a task waits for are its upstream. When Airflow skips a task because an upstream task didn't succeed, it gives it the status upstream_failed. The rule: before you run a task, look at the status of each task it waits for. If any of them is not "success" (it is "failed", or itself "upstream_failed"), mark the task "upstream_failed" and don't call its function. On another pipeline:
download → failed unzip → upstream_failed (waits for download) load → upstream_failed (waits for unzip) make_logo → success (waits for nothing)
Two Python tools help. status.get(name) reads a key like status[name] does, but gives None when the key isn't there yet, where status[name] raises a KeyError. continue ends this turn of a loop and jumps to the next one:
status = {"download": "failed"}
print(status.get("unzip")) # None
for n in [1, 2, 3]:
if n == 2:
continue
print(n) # prints 1, then 3
In this lesson's code, clean_orders raises ValueError("quantity is not a number: two"). Right now the runner prints the error but stores no status for it, and build_report and send_email run anyway. Your finished code must print:
extract_orders done extract_customers done clean_orders failed: quantity is not a number: two extract_orders: success extract_customers: success clean_orders: failed build_report: upstream_failed send_email: upstream_failed
Then the check makes extract_customers fail instead, while clean_orders works. clean_orders must end success, and build_report and send_email must end upstream_failed again: build_report waits for both.
Your task: in run_pipeline, store "failed" for a task that raises. Then, at the top of the run-order loop, mark the task "upstream_failed" if any task it waits for didn't succeed, and skip it with continue.
Assignment
Follow the Coding Tutorial to make run_pipeline record failed tasks and skip every task whose upstream didn't succeed.
Hint
Compare with != "success", not == "failed": send_email waits for build_report, which is upstream_failed, not failed. And marking isn't enough: without continue, the try block still calls the task and writes "success" over the mark.
Free plan: 30 tutor messages. PRO is unlimited.
