Airflow动态任务映射UI仅显示任务总数，求调试特定错误的方案

一、给映射任务加明确标识与日志输出 #

在定义映射任务时，把每个实例处理的参数直接打印到日志里，哪怕UI只显示总数，也能通过日志快速定位出错的任务实例。比如用装饰器写法的任务：

@task
def process_item(item):
    # 先打印当前处理的参数，日志里会直接显示
    print(f"=== 开始处理实例：{item} ===")
    # 你的业务逻辑
    if item == "异常参数":
        raise ValueError(f"{item} 处理失败")

触发任务后，直接查看失败任务的日志，第一行就能看到对应的参数，不用猜测是哪个实例出问题。

二、用CLI工具精准定位错误任务 #

Airflow的命令行工具可以直接筛选和查看特定映射实例的信息：

• 列出某个DAG下所有失败的任务实例：

  airflow tasks list-instances -d 你的DAG_ID -s 起始日期 -e 结束日期 -f failed
  

• 查看指定map索引的任务日志（map索引对应你expand时输入列表的下标，从0开始）：

  airflow tasks logs 你的DAG_ID 任务ID 执行日期 --map-index 1
  

比如你用process_item.expand(item=["a","b","c"])，b对应的map索引是1，直接查这个索引的日志就能快速定位。

三、升级Airflow版本优化UI体验 #

Airflow 2.3.0之后的版本（比如2.4及以上）对动态任务映射的UI做了升级，支持点击任务总数展开查看每个实例的详情，包括状态、对应的参数和日志入口，直接在UI里就能找到出错的实例。如果你的环境允许升级，这是最直接的解决办法。

四、用XCom收集失败实例信息 #

可以在映射任务里，把失败的参数推送到XCom，再用一个汇总任务收集这些信息，方便统一查看：

@task
def process_item(item, ti):
    try:
        # 业务逻辑
        pass
    except Exception as e:
        # 失败时把参数推到XCom
        ti.xcom_push(key="failed_item", value=item)
        raise e

@task
def summarize_failures(ti):
    # 拉取所有process_item任务的XCom数据
    failed_items = ti.xcom_pull(task_ids="process_item", key="failed_item")
    # 过滤掉成功任务的None值
    failed_items = [item for item in failed_items if item is not None]
    print(f"本次执行失败的实例参数：{failed_items}")

# 关联任务
items = generate_items()
process_tasks = process_item.expand(item=items)
summarize_failures()

这样在汇总任务的日志里就能直接看到所有失败的参数，不用逐个去查日志。

内容的提问来源于stack exchange，提问作者Nicolò Gasparini