Airflow中如何将XCom值作为全局变量在Operator外部使用

你提到的shape_change_tables = ti.xcom_pull(key='shape_change_tables', task_ids='push_to_xcom')语句无法直接在Operator外部（DAG文件顶层代码区域）执行，原因是ti是TaskInstance（任务实例）对象，只有在DAG实际运行、对应任务被调度时才会被实例化，而DAG文件的顶层代码是Airflow Scheduler定期解析DAG时执行的，此时DAG还未触发运行，不存在ti对象，也没有生成对应的XCom数据，直接执行会报错。

针对不同的使用场景，有对应的可行解决方案： #

场景1：循环逻辑写在Operator的执行函数内部 #

这种场景可以正常使用XCom拉取，仅需要满足两个前提：

• 执行函数已拿到上下文注入的ti对象：Airflow 2.x使用@task装饰器默认开启上下文注入，Airflow 1.x需要给Operator设置provide_context=True参数
• 拉取XCom的任务的优先级晚于push_to_xcom任务，即设置好上下游依赖：push_to_xcom_task >> 拉取XCom的任务

示例代码：

from airflow.decorators import task

@task
def handle_table_logic(**context):
    ti = context["ti"]
    # 拉取XCom变量
    shape_change_tables = ti.xcom_pull(key="shape_change_tables", task_ids="push_to_xcom")
    ss_cd = ti.xcom_pull(key="ss_cd", task_ids="push_to_xcom")
    env = ti.xcom_pull(key="env", task_ids="push_to_xcom")

    # 你的业务逻辑
    import json
    table_conf = json.load(open("somejson.json", "r"))["tables"]
    for i in table_conf.keys():
        if i in {val for dic in shape_change_tables for val in dic.values()}:
            # 后续处理逻辑
            pass

依赖设置：

# 假设push_to_xcom对应的任务实例名为push_task
push_task >> handle_table_logic()

场景2：循环逻辑写在DAG顶层，用于动态生成Operator #

这种场景不能依赖XCom传值，因为XCom是运行时产物，DAG解析阶段无法获取运行时的参数，可选两种方案：

方案A（推荐）：使用Airflow 2.3+ 动态任务映射 #

动态任务映射不需要在DAG解析阶段提前知道参数值，运行时会自动根据传入的参数生成对应数量的任务，避免解析阶段拿不到参数的问题：

from airflow.decorators import task

@task
def process_single_table(table_name, ss_cd, env):
    # 单表处理逻辑
    print(f"当前环境：{env}, 处理表：{table_name}")

@task
def get_table_list(**context):
    ti = context["ti"]
    shape_change_tables = ti.xcom_pull(key="shape_change_tables", task_ids="push_to_xcom")
    # 提取所有需要处理的表名
    return [val for dic in shape_change_tables for val in dic.values()]

@task
def get_ss_cd(**context):
    return context["ti"].xcom_pull(key="ss_cd", task_ids="push_to_xcom")

@task
def get_env(**context):
    return context["ti"].xcom_pull(key="env", task_ids="push_to_xcom")

# 动态生成多个单表处理任务
push_task >> process_single_table.expand(
    table_name=get_table_list(),
    ss_cd=get_ss_cd(),
    env=get_env()
)

方案B：直接读取dag_run.conf（兼容旧版本Airflow） #

不需要走XCom中转，直接读取触发时传入的conf参数，需要加默认值避免解析阶段报错：

# 解析阶段设置默认值，避免报错
default_shape = []
default_ss_cd = {}
default_env = "dev"
if "dag_run" in locals():
    shape_change_tables = dag_run.conf.get("shape", default_shape)
    ss_cd = dag_run.conf.get("ss_cd", default_ss_cd)
    env = dag_run.conf.get("env", default_env)
else:
    shape_change_tables = default_shape
    ss_cd = default_ss_cd
    env = default_env

# 直接循环生成Operator
import json
table_conf = json.load(open("somejson.json", "r"))["tables"]
for i in table_conf.keys():
    if i in {val for dic in shape_change_tables for val in dic.values()}:
        # 生成对应处理任务
        ...

注意此方案可能存在任务残留问题，即之前触发DAG生成的任务会保留在DAG结构中，仅适合参数变化较小的场景使用。

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