Airflow DAG中LivyOperator无法读取上游PythonOperator推送的XCOM值，报错连接未定义

我来帮你分析这个问题，你遇到的核心问题是LivyOperator的livy_conn_id参数默认不支持Jinja模板渲染，导致Airflow没有解析你写的{{ task_instance.xcom_pull(...) }}模板字符串，而是直接把整个字符串当作连接ID来查找，自然会报错“该连接未定义”。

下面我拆解问题细节并给出解决方案：

问题根源拆解 #

1. 参数模板化支持限制
   不是所有Operator的参数都默认支持Jinja模板渲染，LivyOperator的livy_conn_id就不在默认的template_fields列表里。这意味着Airflow不会处理这个参数里的模板语法，直接原封不动地使用字符串内容。

2. 代码中的语法错误
   你的DAG定义里有几处逗号缺失的问题，这些会导致DAG无法正常加载，必须先修正：

   • default_args里的start_date': dt.datetime(2021, 9, 1) 末尾缺少逗号
   • 'email':['my@email'] 末尾缺少逗号
   • 'email_on_failure': True 末尾缺少逗号
   • PythonOperator的provide_context = True后面也缺少逗号

3. 冗余/错误的代码细节

   • read_vars_func里给livy_conn_id赋值时，test和非test环境用了同一个值，这应该是笔误，建议区分开不同环境的连接ID
   • Airflow 2.x中provide_context=True已经被弃用，推荐直接通过函数参数接收ti（你已经这么做了，没问题，但可以去掉provide_context参数）

解决方案 #

针对LivyOperator无法读取XCOM的问题，推荐使用Airflow 2.x的XComArg特性来直接引用上游任务的XCOM输出，这样不需要依赖Jinja模板，还能自动维护任务依赖，是更简洁的实现方式。

以下是修正后的完整DAG代码：

import datetime as dt
from airflow import DAG
from airflow.operators.dummy import DummyOperator
from airflow.operators.python import PythonOperator
from airflow.providers.apache.livy.operators.livy import LivyOperator

# 定义业务参数
args = {
    'start_month': '20210901',
    'end_month': '20211001',
    'debug': 'True',
    'env': 'test'
}

default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'start_date': dt.datetime(2021, 9, 1),
    'email': ['my@email'],
    'email_on_failure': True,
    'email_on_retry': False,
    'retries': 0,
    'retry_delay': dt.timedelta(minutes=20),
}

dag = DAG(
    dag_id='f_dag',
    default_args=default_args,
    schedule_interval=None,
    catchup=False,
    tags=['DIT'],
    description='Rides'
)

start_task = DummyOperator(task_id='start', dag=dag)
end_task = DummyOperator(task_id='end', dag=dag)

def read_vars_func(ti=None):
    # 修正环境区分的笔误，区分测试和生产环境的Livy连接
    args['livy_conn_id'] = 'livy_test_conn' if args['env'] == 'test' else 'livy_prod_conn'
    args['script_path'] = '/data/src'
    for key, val in args.items():
        ti.xcom_push(key=key, value=val)

read_variables = PythonOperator(
    task_id='read_vars',
    dag=dag,
    # Airflow 2.x中无需provide_context，直接通过参数接收ti
    python_callable=read_vars_func,
)

# 使用XComArg直接引用上游任务的XCOM输出
make_table = LivyOperator(
    task_id='make_table',
    dag=dag,
    livy_conn_id=read_variables.output['livy_conn_id'],
    polling_interval=30,
    name=f"dit_{read_variables.output['start_month']}",
    file=f"{read_variables.output['script_path']}/exec_file.py",
    args=[
        '--start_month', read_variables.output['start_month'],
        '--end_month', read_variables.output['end_month']
    ]
)

start_task >> read_variables >> make_table >> end_task

备选方案（自定义模板支持） #

如果你仍想使用Jinja模板的方式，可以自定义LivyOperator，将livy_conn_id添加到template_fields中，让Airflow对该参数进行模板渲染：

from airflow.providers.apache.livy.operators.livy import LivyOperator

class TemplateLivyOperator(LivyOperator):
    # 将livy_conn_id加入模板字段列表
    template_fields = (*LivyOperator.template_fields, 'livy_conn_id')

之后在DAG中使用TemplateLivyOperator替代原有的LivyOperator即可。不过这种方式不如XComArg简洁，推荐优先使用XComArg方案。

备注：内容来源于stack exchange，提问作者Diana Oryol