Add AssetAndTimeSchedule to schedule DAGs

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) 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 ExternalTaskSensors 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

I agree to follow this project's Code of Conduct

Hi, I'd like to work on this issue, if that's okay :)

Hi, I've purposed a first version of AssetAndTimeSchedule which supports and logic for assets and cron. I think we can implement further functions later, like execution_date, after this PR is accepted.

Please let me know if you have any thoughts :)

Thanks for taking the time to propose a draft! I think it looks good with regard to how the Asset behaves currently, in my humble opinion.

Something that I tried to express in the issue's description is that, given my use case, it would be so nice if the Asset(s) and the CRON expression are "matching" in the sense that a CRON expression is defined by its datetime value (very restrictive) while an Asset does not follow the same rules. For example, a parent DAG having a logical date value being 2 years ago can generate an Asset event which would consumed by the child DAG no matter what. It would be so nice if the child DAG could be triggered only if the CRON expression is matched, and the Asset(s) it is waiting for have been generated by DAGs having a similar logical_date value.

I hope I am clear and I know this is a bit convoluted, but this would be so nice to have. I guess such a thing would require to change the logic of the Asset object consumption somehow.

That being said, thank you very much for the proposal, and I guess your proposal makes sense given how Asset objects are working as of now!

@nailo2c We are unassigning you from this issue as part of our updated assignment policy.

This is not meant to discourage your contribution — quite the opposite! You are still very welcome to work on this issue and submit a PR for it. Simply comment that you are working on it and open a PR when ready.

We found that formal assignments were not working well, as they often prevented others from contributing when the assignee was not actively working on the issue.

Add AssetAndTimeSchedule timetable for time + asset gated scheduling