Airflow 2.9动态任务映射：如何实现运行中任务ID正确映射？

你遇到的问题根源是map_index_template属于任务执行完成后才会渲染的模板逻辑，所以运行中的子任务只能显示默认的索引标识（如generate_tasks[0]），完成后才会替换为模板值。要让子任务在运行时就显示正确名称，需改用map_key参数——它会在动态任务映射阶段就确定每个子任务的标识。

修改后的代码 #

import time
from airflow.operators.python import PythonOperator

def get_task_names():
    task_list = [['task_1'], ['task_2'], ['task_3']]
    return task_list

def generate_tasks(task_name):
    time.sleep(10)

get_task_names = PythonOperator(
    task_id='get_task_names',
    python_callable=get_task_names
)

generate_tasks = PythonOperator.partial(
    task_id='generate_tasks',
    python_callable=generate_tasks,
    # 从op_args中提取任务名称作为子任务标识
    map_key=lambda args: args[0]
).expand(op_args=get_task_names.output)

get_task_names >> generate_tasks

原理说明 #

• map_key接受一个可调用对象，参数是expand传入的每个元素（即task_list里的子列表，比如['task_1']），返回值会作为子任务的标识后缀。
• 这个标识在动态任务创建阶段就会被确定，因此子任务启动后会直接显示generate_tasks[task_1]这类正确名称，无需等待任务完成。

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