Data aware scheduling and dynamic task mapping

[pdavidovicmsft]

Hey folks,

I'd like to utilize data aware scheduling to trigger downstream DAG based on upstream DAG files updates.

It seems like a perfect feature for my use case, but I encountered some issues. Essentially I have dynamic task mapping and if I want to provide outlets as a parameter in set I am sending through .expand_kwargs , I am getting following error: unmappable or already specified argument: outlets.

Then, I tried to pass outlets in .partial method and it works perfectly when values are hardcoded, but if I try to create them dynamically it doesn't work again (my understanding is that arguments which we're sending in .partial method should not be computed in runtime). However, this works just for map_index_template (e.g. map_index_template="{{task.script_args['--provider_name']}}").

Does somebody know how this should be done or it's just a limitation of Airflow?

Additional info: I am using GlueJobOperator, and here is the code:
provider_ingest_jobs = GlueJobOperator.partial(
....
verbose=True,
update_config=True,
create_job_kwargs={
"GlueVersion": "4.0",
"NumberOfWorkers": 2,
"WorkerType": "G.1X",
"Timeout": 300,
},
map_index_template="{{task.script_args['--provider_name']}}",
outlets = datasets,
).expand_kwargs(provider_ingest_mappings)

provider_ingest_mappings is computed in separated task and it's essentially returning a list of tasks.

[potiuk]

I am not sure, but it seems you can't produce datasets with mapped tasks. As I understand that (@uranusjr @cmarteepants ?) the mapped tasks are literally a multiple instances of the same task, and they cannot individually produce dataset each.

You could likely have one other task that depends on those mapped tasks (taking their outputs in) and produce multiple datasets as single set of outlets.

But I am not 100% sure. If I am right, this might be a feature to add in Airflow 3 where we improve dataset support - but again - I am not a big expert on that, that's what I tink is happening here.

[pdavidovicmsft]

Hey @potiuk , thanks for help.

What do you mean by doing that in separate task ? It seems to me I'll encounter same issue I'll need to dynamically compute outlets array in runtime which is not allowed ?

[potiuk]

Datasets are generelly part of the DAG structure, so yes you could not have a "changing" number of those but you could trigger existing datasets via API (in Airflow 2.10 I think maybe even in 2.9) - so you could potentially have task that waits for all the mapped tasks and then triggers the datasets via API. Or maybe even trigger them from the mapped tasks.

[pdavidovicmsft]

Can you point me to some resources? I am not sure that I understand this

Essentially, I'd like to dynamically decide which datasets I'd like to trigger based on DAG config parameters