Apache Airflow动态映射任务XCOM pull返回None问题及优化咨询

问题场景 #

在动态映射的Task Group场景下，BashOperator的xcom_pull调用始终返回None，但动态映射外的XCOM取值正常。代码片段如下：

with DAG(
  params={
    "shas": Param(
      type="array",
      items={
        "type": "string",
      },
    ),
  },
) as dag:
  # Get the input data into the processing chain
  @task(task_id="start", task_display_name="Initial input data")
  def get_input_data(**context) -> list[str]:
    return context["params"]["shas"]

  @task_group(group_id="process_sha")
  def process_sha(sha) -> None:

    @task(task_id="collect_input_data", task_display_name="Prepare Input Data")
    def collect_input_data(sha, **context):
      return sha

    bash_task = BashOperator(
      task_id="bash_task",
      map_index_template="{{ task_instance.xcom_pull(task_ids='collect_input_data') }}",            
      bash_command=r"""
        echo "{{ task_instance.xcom_pull('task_ids=collect_input_data') }}"
      """,
     )

     collect_input_data(sha) >> bash_task


  input_data = get_input_data()
  process_sha.expand(sha=input_data)

问题1：如何正确将sha值传入BashOperator？ #

动态映射的任务实例会带有map_index标识（对应输入数组的索引），原代码中xcom_pull未指定对应实例的索引，导致无法定位到正确的XCOM记录。修正方式如下：

方案1：指定map_index拉取XCOM #

在xcom_pull中添加map_index参数，指向当前任务实例的索引：

bash_task = BashOperator(
  task_id="bash_task",
  bash_command=r"""
    echo "{{ task_instance.xcom_pull(task_ids='collect_input_data', map_index=task_instance.map_index) }}"
  """,
)

方案2：使用短模板变量简化写法 #

Airflow支持用ti替代task_instance，代码更简洁：

bash_task = BashOperator(
  task_id="bash_task",
  bash_command=r"""
    echo "{{ ti.xcom_pull(task_ids='collect_input_data', map_index=ti.map_index) }}"
  """,
)

问题2：是否可无需额外Python任务collect_input_data，直接访问任务组传入的sha参数？ #

完全可以，无需中间过渡任务，有两种直接获取的方式：

方案1：从根任务直接拉取对应索引的sha值 #

利用动态映射的map_index，直接从start任务的XCOM中拉取对应位置的sha：

@task_group(group_id="process_sha")
def process_sha() -> None:  # 移除多余的sha参数
    bash_task = BashOperator(
      task_id="bash_task",
      bash_command=r"""
        echo "{{ ti.xcom_pull(task_ids='start', map_index=ti.map_index) }}"
      """,
    )

input_data = get_input_data()
process_sha.expand()  # 无需传递sha参数

方案2：直接访问动态映射的参数 #

如果保留Task Group的sha参数，可通过Airflow的模板变量{{ params.sha }}直接访问（适用于Airflow 2.3+版本）：

@task_group(group_id="process_sha")
def process_sha(sha) -> None:
    bash_task = BashOperator(
      task_id="bash_task",
      bash_command=r"""
        echo "{{ params.sha }}"
      """,
      params={"sha": sha},  # 将sha传入operator的params字典
    )

input_data = get_input_data()
process_sha.expand(sha=input_data)

内容的提问来源于stack exchange，提问作者Andre-Marcel Hellmund