GitHub user pykenny added a comment to the discussion: Dynamic mapped tasks
group arguments are interpreted as MappedArgument when provided to classic
operators
XCom is designed to pass message between tasks, not task groups or between
tasks and task groups. So based on this design decision, data
serialization/deserialization should occur within a task's scope instead of a
task group's. As the documentation mentioned above says, an operator's XCom
output is just a "reference", whose actual value is not determined until
downstream task completes pulling upstream XCom.
Rather than turning the original operator into a decorated Python operator and
call subprocess within implementation, there are two other workarounds I can
think of:
1. Use templates so that they're evaluated by rendering the template:
```python
bash_template = f"""echo \\
{{
ti.xcom_pull(task_ids='list_dict_generator',map_indexes=ti.map_index)['project']
}}.\\
{{
ti.xcom_pull(task_ids='list_dict_generator',map_indexes=ti.map_index)['dataset']
}}.\\
{{
ti.xcom_pull(task_ids='list_dict_generator',map_indexes=ti.map_index)['table_name']
}}$\\
{{
ti.xcom_pull(task_ids='list_dict_generator',map_indexes=ti.map_index)['partition_id']
}}"""
BashOperator(
task_id="bash_task",
bash_command=bash_template,
env={"MY_VAR": "Hello World"}
)
```
2. Adding an upstream decorated task to generate the command, then pass the
output directly to
```python
@task(task_id="emit_command")
def emit_command(project, dataset, table_name, partition_id) -> str:
return f"echo {project}.{dataset}.{table_name}${partition_id}"
BashOperator(
task_id="bash_task",
bash_command=emit_command(),
env={"MY_VAR": "Hello World"}
)
```
GitHub link:
https://github.com/apache/airflow/discussions/40728#discussioncomment-12100714
----
This is an automatically sent email for [email protected].
To unsubscribe, please send an email to: [email protected]
