A reporting Dag runs at 07:00 and assumes the orders and payments Dags have finished by then. Some days they have not, and the report is built on yesterday's data. How do you make it wait for both?

Schedule the reporting Dag on data, not the clock. In Airflow, the tasks that write orders and payments declare an asset in their outlets, and the reporting Dag sets schedule=[orders, payments]. Airflow marks an asset updated only when the producing task succeeds, and with 2 assets the reporting Dag runs once both have been updated since its last run. A failed or late load means no report, rather than a report on old data. A sensor waiting on the other Dags also works, but in poke mode it holds a worker slot for the whole wait.

1. The producers

from airflow.sdk import DAG, Asset, task

orders = Asset("s3://lake/orders")
payments = Asset("s3://lake/payments")

with DAG(dag_id="load_orders", schedule="@daily", ...):
    @task(outlets=[orders])
    def write_orders(): ...
    write_orders()

with DAG(dag_id="load_payments", schedule="@daily", ...):
    @task(outlets=[payments])
    def write_payments(): ...
    write_payments()

2. The consumer

with DAG(dag_id="daily_report", schedule=[orders, payments], ...):
    @task
    def build_report(): ...
    build_report()

3. What happens on a bad day

The report never runs on half-fresh data, and when it starts late, the delay shows where the problem was.

4. The options compared

5. Points to raise

• An asset updates only on success: a skipped or failed task does not count.
• The asset name is a URI that names the data, so a Dag outside the team can depend on it without knowing which Dag writes it.
• Keep an alert for "no report by 10:00", since an event-driven Dag that never starts makes no noise of its own.

What interviewers want: scheduling on data with assets and outlets, an update only on success, a Dag on several assets waiting for all of them, the problem with a fixed start time, the sensor alternative and its cost, and an alert for a report that never starts.