Apache Airflow XCom跨任务取值失败问题排查求助

原因分析 #

• XCom上下文日期不匹配：尽管两个任务在同一DAG中顺序执行，但如果validate_delimiter_and_load_csv的execution_date与create_dirs_task的日期不一致（比如手动触发时指定了不同日期、或存在DAG catchup遗留实例），会导致拉取到错误的XCom条目。
• Airflow版本兼容性问题：Airflow 2.x之后，provide_context=True已被标记为废弃，依赖该参数传递的ti对象可能存在上下文绑定异常。
• XCom拉取参数模糊：默认xcom_pull会拉取当前任务execution_date对应的XCom，但如果存在同名任务跨DAG或跨execution_date的情况，可能拉取到非目标值。

解决方法 #

1. 明确指定XCom拉取的execution_date #

在拉取XCom时显式传入当前任务的execution_date，确保与推送任务的日期完全匹配：

def validate_delimiter_and_load_csv(ti, **kwargs):
    execution_date = kwargs['execution_date']
    task_logger.info(f"Current execution date: {execution_date}")
    zip_path = ti.xcom_pull(
        task_ids='create_dirs_task',
        key='sftp_raw_data',
        execution_date=execution_date
    )
    output_dir = ti.xcom_pull(
        task_ids='create_dirs_task',
        key='validate_delimiter',
        execution_date=execution_date
    )
    # 后续业务逻辑

2. 改用函数返回值推送XCom（推荐） #

Airflow的PythonOperator默认会将函数返回值以return_value为key推送到XCom，这种方式更简洁且不易出错：

# 修改创建目录的函数，返回路径字典
def create_folders_with_subfolders(base_path, **kwargs):
    execution_date = kwargs['execution_date']
    dir_name = execution_date.strftime("%Y%m%d")
    main_folder = Path(base_path) / dir_name
    main_folder.mkdir(parents=True, exist_ok=True)

    subfolders = ['sftp_raw_data', 'validate_delimiter', 'lowercase', 'data_quality', 'process_files']
    paths = {}
    for subfolder in subfolders:
        subfolder_path = main_folder / subfolder
        subfolder_path.mkdir(exist_ok=True)
        paths[subfolder] = str(subfolder_path)
        task_logger.info(f"Created folder: {str(subfolder_path)}")
    task_logger.info(f"paths {paths}")
    return paths  # 自动推送到XCom

# 修改验证函数，直接拉取返回的字典
def validate_delimiter_and_load_csv(ti, **kwargs):
    paths = ti.xcom_pull(task_ids='create_dirs_task', key='return_value')
    zip_path = paths.get('sftp_raw_data')
    output_dir = paths.get('validate_delimiter')
    task_logger.info(f"zip_path pulled from XCom: {zip_path}")
    task_logger.info(f"output_dir pulled from XCom: {output_dir}")

3. 适配Airflow 2.x的参数传递方式 #

Airflow 2.x推荐直接通过参数接收ti对象，无需依赖provide_context=True，可以移除该参数避免上下文异常：

# 任务定义时移除provide_context=True
create_dirs_task = PythonOperator(
    task_id="create_dirs_task",
    python_callable=create_folders_with_subfolders,
    op_kwargs={'base_path': '/path/to/directory'},
)

validate_task = PythonOperator(
    task_id="validate_delimiter_and_load_csv",
    python_callable=validate_delimiter_and_load_csv,
)

create_dirs_task >> validate_task

排查步骤 #

• 在validate_delimiter_and_load_csv中打印ti.execution_date，与create_dirs_task日志中的execution_date对比，确认日期一致。
• 在Airflow UI的XCom页面，筛选对应dag_id、task_id和execution_date，确认目标key的XCom值存在且正确。

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