apache / apache/airflow

Add `AssetAndTimeSchedule` to schedule DAGs

Open
#58,056 6 comments 4 reactions 0 assignees View on GitHub
area:core kind:feature needs-triage
Dominant language
Python
Stars
46.9k
Forks
17.8k
Avg merge
2d 7h
Merged PRs (30d)
484

Description

### Description

Add a `AssetAndTimeSchedule` timetable so that a DAG is scheduled based on time (using a CRON) but should only start its work once all the Assets it needs are present.

It should be similar to the existing `AssetOrTimeSchedule` timetable that currently existing, but with `AND` instead of `OR`.

### Use case/motivation

As of now in Airflow 3.1.2, one can schedule DAGs based on time (via CRON) or on events (via Assets) or both (via `AssetOrTimeSchedule`).

From my understanding, `AssetOrTimeSchedule` ([doc here](https://airflow.apache.org/docs/apache-airflow/stable/authoring-and-scheduling/timetable.html#asset-timetable-section)) is used to schedule a DAG to run following a CRON but the DAG can also run if some Assets' events of interest have been emitted.

---

What I like about the Asset-Aware scheduling approach is that the dependencies between DAGs are clear. We have a view to see the big picture of all DAG dependencies being connected. Using Asset scheduling allows me to isolate processes into smaller and specific DAGs and connect these DAGs.

What I like about the CRON scheduling approach is that I am sure that my DAG will be scheduled if I use the CRON approach.

---

I am using Airflow to run ETL jobs. I use the medallion architecture to create my tables in my lakehouse, so I have three layers (bronze, silver, gold). I organized my Airflow DAGs as follow:
- Each table has a dedicated DAG per layer, this means that I will have at least a Bronze DAG (for bronze table) and a Silver DAG (for silver table) for a typical table. Each DAG emits an Asset with the name of the table that was modified after the ETL job succeeds.
- A bronze DAG is scheduled using a CRON, since I want to retrieve the data after a particular time. An Asset event is emitted once the DAG succeeds.
- A silver DAG is scheduled using Asset and it listens to the Asset of the Bronze DAG. This is convenient in that I know that the Silver DAG will run automatically after the Bronze DAG succeeded.
- Gold DAGs follow the same logic as Silver DAGs, they are scheduled using Assets from one or many Silver DAGs (whether the Gold table is using many Silver tables or not).

In my case, a Gold table is supposed to be updated everyday. So, if the Gold DAG did not run today, it means that one of its dependencies (Silver DAGs) failed or did not run because one of their dependencies failed (Bronze DAGs). But it is difficult to know about this when using Asset scheduling, in the sense that it is easy to know that a Silver DAG failed, but it is not easy to know that a Gold DAG that depends on this Silver DAG didn't run, at least from a data consumer perspective (they only care about or have vision on the Gold table).

This is why I think that having a `AssetAndTimeSchedule` object to schedule DAGs would be nice. It would let me schedule my Silver and Gold DAGs using a CRON, but the DAG should only start once all the Assets it needs have been emitted. That way, it is easy to define a SLA and/or to make the DAG timeouts if it is running since X hours, so a process can kick in to inform data consumers that a table is not up-to-date. It would also allow the Asset view in the Airflow UI to represent all the dependencies between DAGs.

Right now as an alternative: I am scheduling my DAGs using CRON and the first tasks are `ExternalTaskSensor`s to wait for the parent DAGs to succeed before continuing. That way I am sure that the Gold DAG will run on a particular day, and its ETL job will run once the Silver DAGs it depends on succeeded.

Something that might be a bit challenging with Asset-aware scheduling is that if I clear/backfill a past DagRun then a new Asset will be emitted, and that can potentially trigger another DAG which depends on this Asset. This is why I think it would be nice to have an option to compare the `execution_date` value of the DAG and the `execution_date` of the Assets: if they match then we acknowledge the Assets and the DAG continue, if they don't match then the Assets are not acknowledged.

What do you think about this?

### Related issues

_No response_

### Are you willing to submit a PR?

- [ ] Yes I am willing to submit a PR!

### Code of Conduct

- [x] I agree to follow this project's [Code of Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)

Contributor guide

Open the contributing guide

Research direction

Start with the existing AssetOrTimeSchedule timetable and its documentation in the asset timetable section linked in the issue. Clarify whether the requested scope includes only AND semantics or also matching asset and DAG execution dates. Done should include an agreed timetable design, implementation, and coverage for scheduled runs, asset availability, and backfill or clear behavior.

Written by the indexing model from the issue text.

Assessment

Tech stack
python
Domain
backend, data-engineering
Issue type
Feature
Difficulty
5/5
Estimated time
Over a week
Activity status
Active
Clarity
Needs clarification
Newbie friendliness
38/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.