Expand dataset (asset) scheduling to tasks as well, changing the dag run to a deferred state while it awaits for events triggered externally

[FelipeRamos-neuro]

In the company that I work, we encountered some issues when trying to execute a dag in a manner that could truly support asynchronous execution based on events generated by external services. We devised a strategy using deferrable operators, a table to store state changes between tasks, a sensor to check for state changes and a callback API endpoint for the external services to change the state of the task at event generation. I haven't experimented as much with data-aware scheduling, but as I understand it, right now, it is focused on scheduling and triggering of dag_runs. The idea hinges on introducing some sort of deferred state for dag_runs, where we could, maybe, use the Airflow REST API endpoint for queued events to signal Airflow, from external services, that the dag run can continue execution. I'm not fully aware of how many complexities this could add to the scheduling process, but I think that after the release of Airflow 3.0, this could be a good feature to focus on.

[potiuk]

I think it would have to be specified in a clearer way - likely more concrete examples and discussion in devlist where you could explain what you want to achieve exactly, because your current description is vague and prone to interpretation (I am not sure by reading it what is it really you propose).

Generally if you feel like you can lead such a proposal - anyone can do it - or you can enthuse someone enough to take the leadership, but in order to do it, clear laying out your idea, with examples, reasoning and something that will allow anyone to get enthused is an important prerequisite.

[FelipeRamos-neuro]

Thanks for the feedback, I'll try to be more concise.

[FelipeRamos-neuro]

Today, Airflow supports asynchronous execution through deferrable operators, but it is not entirely async because it still relies on a pull-based mechanism and that requires the dag_run to be on a running state, even if the task is deferred to the triggerer component. The proposal is that we could implement a feature to support a push-based mechanism using assets as dependencies for tasks to be run after a certain event is triggered and "stop" the dag, to a deferred state (if necessary), while it awaits for this external communication. I also just found AIP-82, which attempts to improve external event driven scheduling.

[FelipeRamos-neuro]

A use case would be something like this:
You want to automate the execution of multiple applications with a dag, but all of them work using different resources and maybe even different infrastructures, one is a simple call to an API, the other is an ETL pipeline that runs on external infrastructure provided by a client (that by all accounts we don't necessarily know when is it going to finish it) and the last one is a data validation that we run on our own infrastructure. All sequentially and using previous task outputs for the next task.

In such a case, the dag would have a sensor to monitor the ETL pipeline, but since sensors are traditionally pull-based, it would require the dag_run to be in a state of running and the sensor would also be running indefinitely on the triggerer component. If instead of a sensor, I could define an asset that could receive push-based events from the Airflow REST API, passing for example: dag_id, run_id and task_id as arguments, and set the dag_run to some sort of deferred state until it can resume execution from that push-based event. This would possibly reduce, in certain scenarios, resource usage as it frees worker slots and process inefficiencies such as relying on a pull-based mechanism for monitoring.

[FelipeRamos-neuro]

Airflow already resolves a lot of our issues, in regards to orchestration and automation, but supporting this would also help a lot, especially when dealing with external infrastructure in an asynchronous manner.