Make Dataset URI a templated field

[dhuettenmoser]

Hey All, we're looking to make it possible for some of our DAGs to conditionally execute based on another DAG's completion.
Specifically, we're looking for more of an event based process and don't want to continually update the upstream DAG to add additional conditional DAG triggers (branches), as new downstream DAGs are added. Datasets seem to be a great, simple way to do this, but the addition of a templated URI would make it even better.

For one of our use cases, we have a single task which operates on all raw data for various data sources. When this process completes single or mutiple downstream DAGs should run based on the source. Using a templated field this would involve passing a "source" param into the upstream DAG and setting the outlet of the task to [Dataset("raw_{{ params.source }}")] while downstream dags could be scheduled easily based on the source.

[potiuk]

Sure. I think this ia a sound idea, though implementing it is not trivial. It must be different from the current templating mechanism because JINJA templates are resolved at task execution time. If you would like to work on it - that's cool. If not, then someone might - at some point in time - implement it if they find an interest in it.

[dhuettenmoser]

Thanks for reviewing this! I'd be interested in working on this. I'll clearly have to delve deeper into this, and get a better understanding of the Data aware scheduling mechanisms. My initial, naïve, thought was to use the jinja templates and add a template_fields to the Dataset class, to have it function similar to the File class in the lineage example here but its clear from your notes that's not the correct approach.

[potiuk]

Yes. Template resolving for operators happens in the workers when tasks are about to be executed. But dataset dependencies and scheduling are all done in Scheduler and some of this in the future will be done in Triggerer when data dependencies will start poiling for "real" external resources being changed (currently dataset triggering is limited to internal "producer" > "consumer").

There are a number of implications that should be considered here - for example security. The Scheduler (that will need to resolve those dynamic dataset parts) should never EVER execute a code that DAG author writes. And Jinja2 is essentially code to execute. So Jinja2 is likely - in general - not a good idea to manage those dynamic dataset URLs. We likely need much more declarative way of defining those dynamic naming and something that will not run custom Python code written by DAG authors.

[potiuk]

Another - super important - aspect is performance. Any change here should be very, very carefully tested for performance.

[dhuettenmoser]

I really appreciate all the info and things to consider as part of the implementation. It sounds like some of the future work would cover our use cases as well. I had also seen in an astronomer talk, some possible future changes to the triggerer that would allow for multi-event/reusable triggers, allowing for more complex recurring event based triggering. Until then I'll see what I can come up with for this. Thanks again.