Airflow中ti.xcom_pull()返回None的问题排查与解决求助

一、问题成因排查 #

• 函数未主动返回数据：Airflow的XCom默认仅捕获Python函数的返回值，若get_titanic_data函数没有明确用return返回查询到的数据，XCom中不会存储该任务输出，后续xcom_pull自然返回None。
• XCom大小限制触发丢弃：默认Airflow的XCom有48KB大小限制，若从titanic表查询的数据量超出阈值，Airflow会自动丢弃该XCom记录，导致无法获取数据。
• task_ids参数异常：ti.xcom_pull(task_ids=['get_titanic_data'])传入列表格式时，若存在Airflow版本兼容问题，或任务ID拼写/大小写与定义不一致，会导致无法匹配到对应XCom。
• DAG运行实例隔离：若task_get_titanic_data和task_process_titanic_data不属于同一个DAG运行实例（比如手动触发了不同的DAG Run），xcom_pull无法跨实例拉取数据。
• XCom存储后端异常：若自定义了XCom存储后端（如非默认数据库存储），可能存在存储失败、权限不足等问题，导致数据未持久化。

二、修复方案 #

1. 确保函数返回数据 #

修改get_titanic_data函数，明确返回查询结果：

def get_titanic_data():
    # 原有数据库连接与查询逻辑
    conn = psycopg2.connect(host='localhost', database='your_db', user='your_user', password='your_pw')
    cursor = conn.cursor()
    cursor.execute("SELECT * FROM titanic")
    data = cursor.fetchall()
    conn.close()
    # 必须添加return语句
    return data

2. 处理大数据量场景 #

若数据量超出XCom限制，采用以下方案：

• 将数据写入临时文件（本地CSV或对象存储），在process_titanic_data中直接读取文件，而非通过XCom传递。
• 调整airflow.cfg中的max_xcom_size参数，增大允许的XCom大小（仅适用于数据量略超阈值的场景，不推荐超大数据）。

3. 修正xcom_pull参数 #

• 核对task_ids的拼写和大小写，确保与DAG中定义的task_get_titanic_data完全一致。
• 若无需批量拉取，直接传入字符串格式：ti.xcom_pull(task_ids='get_titanic_data')。

4. 验证DAG运行实例一致性 #

在Airflow UI中查看两个任务的“DAG Run ID”，确保属于同一个运行实例，避免跨实例拉取数据。

5. 检查XCom存储状态 #

• 进入Airflow UI的「Admin > XComs」页面，搜索task_get_titanic_data对应的记录，确认数据是否存在。
• 若使用自定义存储后端，检查后端服务运行状态及权限配置。

修正后DAG代码示例 #

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

def get_titanic_data():
    conn = psycopg2.connect(
        host='localhost',
        database='your_db',
        user='your_user',
        password='your_pw'
    )
    cursor = conn.cursor()
    cursor.execute("SELECT * FROM titanic")
    titanic_data = cursor.fetchall()
    cursor.close()
    conn.close()
    # 明确返回查询数据
    return titanic_data

def process_titanic_data(ti):
    data = ti.xcom_pull(task_ids='task_get_titanic_data')
    if not data:
        raise Exception("No data")
    # 后续数据处理逻辑
    print(f"成功获取{len(data)}条泰坦尼克号数据")

with DAG(
    'titanic_pipeline',
    start_date=datetime(2024, 1, 1),
    schedule_interval='@daily',
    catchup=False
) as dag:
    task_get = PythonOperator(
        task_id='task_get_titanic_data',
        python_callable=get_titanic_data
    )

    task_process = PythonOperator(
        task_id='task_process_titanic_data',
        python_callable=process_titanic_data,
        provide_context=True  # Airflow 2.x+也可通过op_kwargs传递ti参数
    )

task_get >> task_process

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