Airflow非PythonOperator任务传值及Jinja模板报错排查

一、Airflow任务间传值的可行方式 #

1. XCom（原生核心方式，支持所有Operator） #

（1）PythonOperator显式推送/拉取

这是最常见的用法，通过任务实例（ti）的xcom_push和xcom_pull方法操作：

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

def push_fusion_args(ti):
    # 以指定key推送值
    ti.xcom_push(key="fusion_args", value={"source": "db", "limit": 100})

def pull_fusion_args(ti):
    # 拉取指定任务、指定key的XCom值
    args = ti.xcom_pull(task_ids="push_args_task", key="fusion_args")
    print(f"获取到的参数：{args}")

with DAG(
    dag_id="xcom_basic_demo",
    start_date=datetime(2024, 1, 1),
    schedule_interval=None
) as dag:
    push_task = PythonOperator(
        task_id="push_args_task",
        python_callable=push_fusion_args
    )

    pull_task = PythonOperator(
        task_id="pull_args_task",
        python_callable=pull_fusion_args
    )

    push_task >> pull_task

（2）非PythonOperator通过Jinja模板拉取

像BashOperator、SqlOperator这类无Python可调用函数的Operator，可直接在支持模板渲染的字段中用Jinja语法拉取XCom，注意引号转义：

from airflow.operators.bash import BashOperator

# 依赖前面的push_args_task任务
bash_pull_task = BashOperator(
    task_id="bash_pull_xcom",
    # 用单引号包裹bash命令，内部双引号不需要转义；反之亦然
    bash_command='echo "从push_args_task获取的参数：{{ ti.xcom_pull(task_ids=\'push_args_task\', key=\'fusion_args\') }}"',
    dag=dag
)

注：需确认Operator的字段是否支持模板渲染，比如BashOperator的bash_command默认支持，若自定义Operator需在template_fields中声明可模板化的字段。

（3）TaskFlow API简化XCom（Airflow 2.0+）

用装饰器定义任务，函数返回值自动作为XCom推送，下游任务直接接收参数，无需手动调用xcom_push/pull：

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

@dag(start_date=datetime(2024, 1, 1), schedule_interval=None)
def taskflow_xcom_demo():
    @task
    def push_args_task():
        # 返回值自动推送到XCom
        return {"fusion_args": {"source": "api", "timeout": 30}}

    @task
    def pull_args_task(pushed_data):
        # 直接接收上游任务的返回值
        print(f"TaskFlow获取的参数：{pushed_data['fusion_args']}")

    data = push_args_task()
    pull_args_task(data)

taskflow_dag = taskflow_xcom_demo()

2. Airflow Variables（全局静态值场景） #

如果是跨DAG复用、长期稳定的配置值，用Variables存储（也可在Airflow UI中管理）：

from airflow.models import Variable
from airflow.operators.python import PythonOperator

# 提前设置变量（UI或代码中设置）
Variable.set("global_fusion_args", '{"region": "cn", "env": "prod"}')

def use_global_args():
    # 读取变量，deserialize_json=True自动转成字典
    args = Variable.get("global_fusion_args", deserialize_json=True)
    print(f"全局参数：{args}")

⚠️ 注意：Variables是全局存储，不适合任务间动态生成的临时值，频繁修改可能引发并发问题。

3. 外部存储（大数据量场景） #

XCom默认限制48KB，若传递大体积数据，可借助外部数据库（如PostgreSQL）或文件系统（如本地文件、S3）：

import json
import psycopg2
from airflow.operators.python import PythonOperator

def save_to_db():
    conn = psycopg2.connect("dbname=airflow user=airflow password=airflow host=postgres")
    cur = conn.cursor()
    # 存储JSON格式的大数据
    cur.execute("INSERT INTO task_params (key, value) VALUES (%s, %s)", 
                ("fusion_args", json.dumps({"data": [i for i in range(1000)]})))
    conn.commit()
    cur.close()
    conn.close()

def read_from_db():
    conn = psycopg2.connect("dbname=airflow user=airflow password=airflow host=postgres")
    cur = conn.cursor()
    cur.execute("SELECT value FROM task_params WHERE key = %s", ("fusion_args",))
    result = json.loads(cur.fetchone()[0])
    print(f"从数据库读取的大数据：{len(result['data'])}条")
    cur.close()
    conn.close()

二、Jinja模板渲染错误的原因及修复 #

你提供的代码'{{ ti.xcom_pull(task_ids="get_fusion_args",key="fusion_args) }}'存在语法错误：key="fusion_args结尾缺少一个双引号，导致Jinja模板引擎无法正确解析参数。

正确写法： #

'{{ ti.xcom_pull(task_ids="get_fusion_args", key="fusion_args") }}'

额外排查点： #

• 前序任务get_fusion_args未成功执行，没有生成对应key的XCom值；
• 前序任务推送XCom时的key与拉取时的fusion_args不匹配；
• 当前使用的Operator字段不支持模板渲染：需确认该字段是否在Operator的template_fields列表中，若不在则无法解析Jinja语法。

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