add option to set DAG level Dataset outlets

[nimrodV81]

Description

add a new argument outlets to the DAG constructor with the same effect as the BaseOperator's outlets argument when the full dag run is successfull

Use case/motivation

use case:
I'm currently maintaining several DAGs with "Dataset Scheduling", where each DAG handles a specific dataset and declares that it is ready for downstream DAGs by an EmptyOperator with an outlets value
this means I need to fan-in all the leafs of each DAG:

suggestion:
add an outlets argument to the DAG constructor to mark the Dataset(s) as ready when the dag run is successful

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

[boring-cyborg]

Thanks for opening your first issue here! Be sure to follow the issue template! If you are willing to raise PR to address this issue please do so, no need to wait for approval.

[RNHTTR]

Initially I thought this was a perfect use case for EmptyOperator, but that's just one more thing to maintain in potentially many DAGs.

I've assigned you to this issue :)

[nimrodV81]

@RNHTTR Thanks for the feedback
I want to add this feature myself, but I'm afraid I need some guidance
I've forked the repo and started to dive into the code
Can you point me to a good place to place on_success code for a DagRun
(similar to airflow.models.taskinstance._run_finished_callback for TaskInstance)

[nimrodV81]

@RNHTTR after some observation, the design I came up with is to create a callback and append it to DAG.on_success_callback during init
is this an acceptable solution?

[RNHTTR]

@RNHTTR after some observation, the design I came up with is to create a callback and append it to DAG.on_success_callback during init is this an acceptable solution?

Interesting! Does this make an API call to trigger a Dataset and thus avoids any changes to the Airflow source code? Or does this take a new parameter in the DAG's initialization?

[nimrodV81]

@RNHTTR to trigger the API call I would need to make some small code changes
the design is to add and outlets argument to DAG initialization (to match the usage of BaseOperator)
then in DAG.__init__, if the user provided Datasets in outlets, create a callback that calls dataset_manager.register_dataset_change and append it to DAG.on_success_callback
how does that sound?

[RNHTTR]

I think it sounds good. What about having on_success_outlets and on_failure_outlets?

What do you think @uranusjr

[uranusjr]

I’m not too sure about on_failure_outlets but an outlet that emits exactly once on success sounds reasonable to me.