如何在Airflow 2.5.3中结合KubernetesPodOperator实现动态任务映射

你的问题出在动态映射（expand）的执行时机与模板渲染时机不匹配：expand是在DAG解析阶段处理的，此时模板字符串{{ ti.xcom_pull(...) }}还未被渲染，会被当作普通字符串按字符拆分，导致每个任务的arguments变成单个字符。

正确解决方案 #

Airflow动态映射支持直接引用上游任务的XCom输出作为映射数据源，无需使用xcom_pull模板。具体有两种实现方式：

方式1：上游返回完整的arguments列表

让generate_list函数直接返回每个任务对应的arguments参数列表，然后在expand中直接引用上游任务的输出：

def generate_list(**kwargs):
    # 每个元素对应一个任务的arguments参数
    return [
        ["--input", "item1"],
        ["--input", "item2"],
        ["--input", "item3"]
    ]

generate_list_task = PythonOperator(
    task_id='generate_list',
    python_callable=generate_list,
    dag=dag
)

# 动态生成KubernetesPodOperator任务
mapped_k8s_task = KubernetesPodOperator.partial(
    dag=dag,
    task_id='my_task',
    name='my_name',
    image='my_image',
    image_pull_policy='IfNotPresent',
    do_xcom_push=True,
    is_delete_operator_pod=True,
    in_cluster=True,
    get_logs=True
).expand(
    arguments=generate_list_task.output
)

# 建立任务依赖
generate_list_task >> mapped_k8s_task

方式2：上游返回单个参数值，在partial中构建arguments

如果上游仅返回需要传递的参数值列表，可通过expand传递单个值，再在partial中用{{ item }}模板（动态映射专属模板变量）构建arguments：

def generate_list(**kwargs):
    # 返回单个参数值的列表
    return ["item1", "item2", "item3"]

generate_list_task = PythonOperator(
    task_id='generate_list',
    python_callable=generate_list,
    dag=dag
)

# 动态生成KubernetesPodOperator任务
mapped_k8s_task = KubernetesPodOperator.partial(
    dag=dag,
    task_id='my_task',
    name='my_name',
    image='my_image',
    image_pull_policy='IfNotPresent',
    do_xcom_push=True,
    is_delete_operator_pod=True,
    in_cluster=True,
    get_logs=True,
    # 用{{ item }}引用动态映射的当前元素
    arguments=["--input", "{{ item }}"]
).expand(
    # 将上游返回的列表作为item的数据源
    item=generate_list_task.output
)

generate_list_task >> mapped_k8s_task

原理说明 #

• Airflow动态映射的expand方法会直接读取上游任务的XCom返回值（通过task.output），无需手动调用xcom_pull。
• 动态映射中的{{ item }}是专属模板变量，会自动替换为当前映射任务对应的列表元素，这个渲染过程是在任务运行阶段完成的，不会出现字符串拆分问题。

内容的提问来源于stack exchange，提问作者Chris