Airflow v2.11中XCom 1024条目限制的解决方案咨询

问题复现 #

在Airflow v2.11中，当任务返回超过1024条目的列表时，会触发XCom的条目限制导致失败。以下是复现代码：

import os, datetime
import json, random, string

from airflow.decorators import dag, task

@dag(dag_id="TestLargeDag", max_active_runs=1, schedule=None, schedule_interval=None, start_date=datetime.datetime(2025, 10, 22),
     catchup=False, tags=["teamCottageLabs", "test"])
def simple_test():
    @task(task_id="get_file_list", retries=0)
    def get_file_list():
        large_list = []
        for i in range(1100):
            large_list.append(random.choices(string.ascii_letters + string.digits, k=8))
        return large_list

    @task(task_id="print_one_string", retries=0)
    def print_one_string(str_to_print):
        print(str_to_print)

    file_list = get_file_list()
    print_one_string.expand(str_to_print=file_list)

simple_test()

失败的尝试：直接通过本地文件传递 #

尝试用本地文件跨任务传递列表，但代码在DAG解析阶段就执行了文件读取操作，此时生成文件的任务还未运行，导致文件不存在而失败：

import os, datetime, json
from airflow.decorators import dag, task

file_name = "/tmp/temp.json"

@dag(dag_id="TestLargeDag", max_active_runs=1, schedule=None, schedule_interval=None, start_date=datetime.datetime(2025, 10, 22),
     catchup=False, tags=["teamCottageLabs", "test"])
def simple_test():

    @task(task_id="get_file_list", retries=0)
    def get_file_list():
        large_list = []
        for i in range(1100):
            large_list.append(random.choices(string.ascii_letters + string.digits, k=8))

        with open(file_name, 'w') as f:
            json.dump(save_list, f, indent=4)

    @task(task_id="print_one_string", retries=0)
    def print_one_string(str_to_print):
        print(str_to_print)

    get_file_list()
    file_list = []
    with open(file_name) as file:
        file_list = json.load(file)

    print_one_string.expand(str_to_print=file_list)

simple_test()

可行的解决方案 #

1. 调整XCom全局条目限制 #

Airflow默认通过max_xcom_entries参数限制单个XCom的条目数（默认1024），可以通过以下方式修改：

• 修改airflow.cfg中的core.max_xcom_entries值，比如设置为2000
• 通过环境变量AIRFLOW__CORE__MAX_XCOM_ENTRIES=2000配置

注意：这是全局配置，会影响所有DAG，需考虑数据库存储压力。

2. 修正本地文件传递方案（适用于单节点Airflow） #

将文件读取操作放在独立任务中，确保在任务运行阶段执行（而非DAG解析阶段），集群环境需替换为共享存储路径（如NFS）：

import os, datetime, json, random, string
from airflow.decorators import dag, task

file_name = "/tmp/temp.json"  # 集群环境替换为共享存储路径

@dag(dag_id="TestLargeDag", max_active_runs=1, schedule=None, start_date=datetime.datetime(2025, 10, 22), catchup=False, tags=["teamCottageLabs", "test"])
def simple_test():
    @task(task_id="get_file_list", retries=0)
    def get_file_list():
        large_list = []
        for i in range(1100):
            large_list.append(''.join(random.choices(string.ascii_letters + string.digits, k=8)))
        with open(file_name, 'w') as f:
            json.dump(large_list, f, indent=4)

    @task(task_id="read_file_list", retries=0)
    def read_file_list():
        with open(file_name) as file:
            return json.load(file)

    @task(task_id="print_one_string", retries=0)
    def print_one_string(str_to_print):
        print(str_to_print)

    get_file_list()
    file_list = read_file_list()
    print_one_string.expand(str_to_print=file_list)

simple_test()

3. 使用Redis作为中间缓存（推荐集群环境） #

Redis可以轻松存储大列表且支持分布式访问，示例代码：

import os, datetime, random, string
import redis
from airflow.decorators import dag, task

# 根据你的Redis实例配置修改
REDIS_HOST = "redis"
REDIS_PORT = 6379
REDIS_DB = 0
REDIS_KEY = "large_file_list"

@dag(dag_id="TestLargeDag", max_active_runs=1, schedule=None, start_date=datetime.datetime(2025, 10, 22), catchup=False, tags=["teamCottageLabs", "test"])
def simple_test():
    @task(task_id="get_file_list", retries=0)
    def get_file_list():
        r = redis.Redis(host=REDIS_HOST, port=REDIS_PORT, db=REDIS_DB)
        large_list = []
        for i in range(1100):
            large_list.append(''.join(random.choices(string.ascii_letters + string.digits, k=8)))
        # 清理旧数据避免干扰
        r.delete(REDIS_KEY)
        # 将列表存入Redis
        r.rpush(REDIS_KEY, *large_list)

    @task(task_id="fetch_list_from_redis", retries=0)
    def fetch_list_from_redis():
        r = redis.Redis(host=REDIS_HOST, port=REDIS_PORT, db=REDIS_DB)
        # 读取全部条目并解码为字符串
        return [item.decode('utf-8') for item in r.lrange(REDIS_KEY, 0, -1)]

    @task(task_id="print_one_string", retries=0)
    def print_one_string(str_to_print):
        print(str_to_print)

    get_file_list()
    file_list = fetch_list_from_redis()
    print_one_string.expand(str_to_print=file_list)

simple_test()

4. 切换XCom后端为Redis/S3 #

Airflow支持自定义XCom后端，将默认的SQL后端替换为Redis或S3，彻底突破条目限制：

• 安装对应provider：pip install apache-airflow-providers-redis
• 修改airflow.cfg中的xcom_backend为airflow.providers.redis.xcom_backend.RedisXComBackend
• 配置Redis连接信息，所有XCom数据将自动存储到Redis，无需修改任务代码

5. 拆分大列表为多个小任务 #

如果不想依赖外部存储，可以将大列表拆分为多个不超过1024条的子列表，分别处理：

import os, datetime, random, string
from airflow.decorators import dag, task

@dag(dag_id="TestLargeDag", max_active_runs=1, schedule=None, start_date=datetime.datetime(2025, 10, 22), catchup=False, tags=["teamCottageLabs", "test"])
def simple_test():
    @task(task_id="get_file_lists", retries=0)
    def get_file_lists():
        large_list = []
        for i in range(1100):
            large_list.append(''.join(random.choices(string.ascii_letters + string.digits, k=8)))
        # 拆分为两个子列表
        return [large_list[:550], large_list[550:]]

    @task(task_id="print_one_string", retries=0)
    def print_one_string(str_to_print):
        print(str_to_print)

    file_lists = get_file_lists()
    # 遍历子列表分别展开
    for sub_list in file_lists:
        print_one_string.expand(str_to_print=sub_list)

simple_test()

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