Dataset Consumer Triggering Producers

[zander1976]

Good Afternoon Everybody,

Is it possible to have a consumer of a dataset block and trigger the producers dataset?

Aggregation Dag:

• Node 1 - Needs File 1 from File 1 Dag
• Node 2 - Needs File 2 from File 2 Dag
• etc

File 1 Dag produces File 1
File 2 Dag produces File 2

Starting the aggregation dag would trigger File 1 dag when it reached node 1 then File 2 dag when it reached node 2. I guess this would be similar to lazy loading classes.

We have tons of files created by mathmatical models and our workflow is often aggregating all the different models into a final report. We could generate each model based on a clock and have the aggregation run when each of the producers are done but it would be great if I could start the final report and have it resolve and trigger it's dependencies.

Thank you,
Ben
ps. If someone could point me to the right place in the code then i might be able to write the solution myself.

[potiuk]

I am not sure if I understand your use case. I have a feeling that you want to mix producer/consumer approach with "traditional" behaviour of Airlfow from before datasets where you would trigger a number of tasks, they'd produce the outputs and then their success would trigger the final DAG. This is just "classic" DAG from Airflow where datasets should not be involved.

But I might be completely misunderstanding you, your description was a bit incomplete or maybe just I miss some assumptions you have in your head.

Maybe try to visualise your approach and explain it better (but I would encourage you to maybe look first if "classic" approach of Airlfow will not fit your needs better).

[zander1976]

Hello Potiuk,

Thank you for your time and help. As you mentioned, a mix of traditional and datasets is what I am looking for.

In the traditional system, for the example below, the main DAG controls the flow of the of file generation and consumption. The problem come in when you want to use a section of the DAG in a different DAG. In the example below, we might have multiple DAGs requiring the course_list so we might end up copying and pasting the code. That might be ok if the course list generation was managed by a single node but often ours are large groups of nodes usually contained in a TaskGroup.

The data driving (dataset) design addresses this problem and allows for separation of concerns (do one thing well). We can create seperate dags for the student list, the course list and the main DAG requiring the two different files. This allows us to use course_list, as an example, to be used in different locations without copying and pasting. The problem with this system is that the control is in the hands of the file producers. Course list and student list run on a schedule and when both complete it starts the main DAG to produce a final aggregated report.

In a hybrid proposal, the main DAG is scheduled and maintains control of the flow while still separating the concerns into separate DAGs. In the below example, the main DAG is scheduled daily. On start, the first node requires a dataset (course_list) that is produced in a different DAG. The system should pause the main DAG, trigger the required create course_list dag identified by the Dataset and wait for the completion. Once, course_list DAG has been completed the control is handled back to the main DAG to progress to node 2 which in this case requires the dataset(student_list)

Once the student list DAG completes then it continues on the produce the final report.

A traditional DAG idea

Main dag(schedule="@daily"
Create course_list.txt -> Create student_list.txt -> count students in degree -> write to database -> generate report

The new Dataset triggering allows for:

Aggregate DAG (schedule=[Dataset(student_list), Dataset(course_list)]
count students in degree -> write to database -> generate report

Create student list Dag (schedule="@daily", outlet=[Dataset(student_list)])
Create student_list.txt

Create course list Dag (schedule="@daily", outlet=[Dataset(course_list)])
Create course_list.txt

Hybrid Pipeline with Datasets

Main dag(schedule="@daily"):
Trigger create course_list DAG and wait -> Trigger create student_list DAG and wait -> count students in degree -> write to database -> generate report

Create student list Dag (outlet=[Dataset(student_list)])
Create student_list.txt

Create course list Dag (outlet=[Dataset(course_list)])
Create course_list.txt

[potiuk]

In the traditional system, for the example below, the main DAG controls the flow of the of file generation and consumption. The problem come in when you want to use a section of the DAG in a different DAG. In the example below, we might have multiple DAGs requiring the course_list so we might end up copying and pasting the code.

What is the problem with sparation of the common code to a utility library and using it in multiple DAGs? DAGs are Python code and keeping things DRY by separating code to commonn libraries is absolutely expected.

We even explain it in https://airflow.apache.org/docs/apache-airflow/stable/modules_management.html#typical-structure-of-packages

[potiuk]

In a hybrid proposal, the main DAG is scheduled and maintains control of the flow while still separating the concerns into separate DAGs. In the below example, the main DAG is scheduled daily. On start, the first node requires a dataset (course_list) that is produced in a different DAG. The system should pause the main DAG, trigger the required create course_list dag identified by the Dataset and wait for the completion. Once, course_list DAG has been completed the control is handled back to the main DAG to progress to node 2 which in this case requires the dataset(student_list)

But IMHO this is entirely not needed. Why do you want to "pause" main DAG if it is actually anyhow waiting for the two lists to run?

You can just run both lists on schedule and have multiple DAGs just waiting for both. Or maybe I am missing something ?

[Daily schedule] Couse DAG -> produces COURSE_LIST
[Daily schedule] Student list  DAG -> produces STUDENT_LIST

STUDENT_LIST -\
                    -- DAG1 consuming both
                    -- DAG2 consuming both
COURSE_LIST  -/

I believe that would do what you want.

[zander1976]

Good Evening,

Thank you again for your help. You are correct the problem can be solved in multiple ways but maybe not in an elegant way. The code can be shared but what about the execution. Let's try thinking from the report backwards. Lets say you have a daily and weekly report. The daily report uses FIles 1, FIles 2 and Files 3 while the weekly only uses Files 1 and 3. We can assume that the creation of the files isn't simple and the consumption of the files is cpu intensive.

Possible solutions:

Option 1 - Daily and Weekly scheduled DAGs with datasets for Files1, 2 and 3:

So lets imagine you could define a DAG that requires a list of datasets and those datasets would refresh automatically before starting the dag. Something like:

Daily Dag (schedule=@daily):
Process Dataset(FIle1) -> Process Dataset(FIle2) -> Process Dataset(FIle1) -> complete report
Weekly Dag (schedule=@Weekly):
Process Dataset(FIle1) -> Process Dataset(FIle3) -> complete report

In this option, the daily and weekly dags would run on a schedule and the files1, 2 and 3 datasets would be triggered automatically. At the end of the week, you might notice that daily and weekly are generating and using the same files at the same time. That isn't a problem because we are sharing both the code and the execution. Both daily and weekly dags have said update FIle 1 and that caused the file 1 dataset producer DAG to run. Daily and weekly dags are both added to the waiting list for file 1 to be ready and the second trigger is simply ignored.

This could be done using shared python code in a traditional dag but you could have multiple dags writing/consuming the same files at the same time. The second idea might be to use dataset for file 1, 2 and 3 that update daily but both daily dag and weekly dag would run daily as the dependencies have updated. That's a lot of extra compute to run a weekly report daily. You could add a 4th schedule dataset called blocker that does nothing but run weekly and attach it to weekly dag. This solves part of the problem but what data files do you have? Each day files 1 and 3 are updates telling the weekly dag that the dependencies are updated. On the 7th day when the weekly blocker dag finally triggers the weekly dag will run immediately but Files 1 and 3 could be from day 6 or day 7 depending on the timing of the blocker dag. As you can see, this option starts getting complicated with tons of scheduled dags. You could create a master dag that triggers all the datasets and they in tern trigger the report but now you are creating a dag that triggers all the dags and a second one for creating the report.

What would be really amazing is being able to think of datasets like legos. I need to create a report that consumes files 1, files 2 and files 3 and generates a report like:

File 1 dag (not scheduled):
extract -> transform -> load

File 2 dag (not scheduled):
extract -> transform -> load

File 3 dag (not scheduled):
extract -> transform -> load

Main scheduled dags:
Daily report (schedule=daily, requires dataset file 1 and file 2):
Weekly report (schedule=weekly, requires datasets file 1, file 2 and file 3):

And know that the datasets will be updated just in time for the required report without collisions or duplicated efforts. You could go one step further and add a stale time out so if file 1 dataset was generated within the hour then skip the update. It could be even better if you only updated when a node actually needs the dataset. Some branches coudl be skipped so why run all the datasets frist.

Thank you,
Ben

[potiuk]

Sure. This is is what is planned some time in the future. If you would like to implement and contribute to it via PRs - you are most welcome.