Airflow中@task.external_python任务间共享Python列表全局变量问题

一、任务间共享Python列表（解决PicklingError） #

Airflow中不能直接用全局变量共享数据，因为任务可能分布在不同Worker节点，全局变量无法跨进程/节点同步。XCOM是官方推荐的跨任务数据传递方式，你之前用XCOM没解决大概率是用法错误，以下是正确实现方式：

正确实现步骤 #

1. 每个任务通过ti.xcom_push()将列表推送到XCOM，或直接return列表（Airflow会自动将return值推送到XCOM）
2. 后续任务通过ti.xcom_pull()获取列表，追加值后再推回XCOM

注意：Airflow 2.x默认用JSON序列化XCOM数据，如果你要传递的列表包含JSON无法序列化的对象（比如自定义类实例），需要：

• 要么将对象转为可序列化的格式（比如字典）
• 要么修改Airflow配置，开启pickle序列化（不推荐，有安全风险）：

  [core]
  enable_xcom_pickling = True
  

示例代码 #

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime

def init_list(ti):
    # 初始化列表并推送到XCOM
    initial_list = ["task1_value"]
    ti.xcom_push(key="shared_list", value=initial_list)

def append_to_list(ti):
    # 从XCOM获取列表
    current_list = ti.xcom_pull(task_ids="init_list", key="shared_list")
    # 追加值
    current_list.append("task2_value")
    # 推回XCOM
    ti.xcom_push(key="shared_list", value=current_list)
    return current_list

def print_final_list(ti):
    final_list = ti.xcom_pull(task_ids="append_to_list", key="shared_list")
    print(f"Final shared list: {final_list}")

default_args = {
    'owner': 'airflow',
    'start_date': datetime(2023, 1, 1),
}

with DAG('shared_list_dag', default_args=default_args, schedule_interval=None) as dag:
    task1 = PythonOperator(
        task_id='init_list',
        python_callable=init_list,
        provide_context=True,
    )

    task2 = PythonOperator(
        task_id='append_to_list',
        python_callable=append_to_list,
        provide_context=True,
    )

    task3 = PythonOperator(
        task_id='print_final_list',
        python_callable=print_final_list,
        provide_context=True,
    )

    task1 >> task2 >> task3

PicklingError原因及解决 #

• 如果你用了全局变量（比如在DAG定义外定义shared_list = []），Airflow在序列化DAG对象时会尝试pickle这个全局变量，如果变量包含不可pickle的内容（比如未序列化的函数引用、自定义类）就会报错。
• 解决方式：完全移除全局变量，改用XCOM传递数据。

二、单个任务失败时继续执行后续任务并标记失败 #

要实现“单个任务失败不阻断后续任务，且失败任务被标记”，需要设置任务的trigger_rule为all_done，该规则表示不管前置任务成功/失败/跳过，当前任务都会执行。

示例代码（结合共享列表需求） #

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime

def init_list(ti):
    initial_list = ["task1_value"]
    ti.xcom_push(key="shared_list", value=initial_list)
    # 模拟任务失败
    # raise Exception("Task1 failed")

def append_to_list(ti):
    current_list = ti.xcom_pull(task_ids="init_list", key="shared_list", default=[])
    current_list.append("task2_value")
    ti.xcom_push(key="shared_list", value=current_list)

def print_final_list(ti):
    final_list = ti.xcom_pull(task_ids="append_to_list", key="shared_list", default=[])
    print(f"Final shared list: {final_list}")

default_args = {
    'owner': 'airflow',
    'start_date': datetime(2023, 1, 1),
}

with DAG('failure_continue_dag', default_args=default_args, schedule_interval=None) as dag:
    task1 = PythonOperator(
        task_id='init_list',
        python_callable=init_list,
        provide_context=True,
    )

    task2 = PythonOperator(
        task_id='append_to_list',
        python_callable=append_to_list,
        provide_context=True,
        trigger_rule='all_done',  # 关键：不管task1状态如何都执行
    )

    task3 = PythonOperator(
        task_id='print_final_list',
        python_callable=print_final_list,
        provide_context=True,
        trigger_rule='all_done',
    )

    task1 >> task2 >> task3

DAG标记失败但任务成功的异常排查 #

这个问题通常是因为：

1. DAG的catchup设置导致历史任务状态影响当前DAG状态
2. 任务的execution_timeout或retries配置异常，导致DAG层面判定失败但任务实际成功
3. Airflow元数据不一致（Docker部署下可能是元数据库同步问题）

解决方式：

• 检查DAG的catchup=False，避免历史任务干扰
• 查看任务的详细日志，确认任务是否真的成功（比如是否有隐性异常被捕获但未抛出）
• 重启Airflow Webserver和Scheduler，或清理元数据库中的无效任务实例记录

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