Cloud Composer Airflow v2.2.5升级v2.5.1后XCom模板报错

问题背景 #

将Cloud Composer中的Airflow版本从v2.2.5升级到v2.5.1后，部分运行中的DAG出现Jinja模板渲染错误，日志如下：

[2023-07-05, 00:40:16 UTC] {abstractoperator.py:609}
ERROR - Exception rendering Jinja template for task 'start_bulk', 
field 'skip_message'. 
Template: "{{ ti.xcom_pull(task_ids=['prevState_'+ti.task_id], key='skip_message')[0] }}"
        
jinja2.exceptions.UndefinedError:
airflow.models.xcom.LazyXComAccess object has no element 0

移除模板中的[0]索引后，DAG即可正常运行。以下是对该现象原因及Airflow相关变更的解析。

相关代码 #

# prevState函数内部逻辑
# 如果需要跳过后续任务则返回False（同时推送XCom）
kwargs['ti'].xcom_push(key='skip_message', value=skip_message)
return False  # 返回False时触发后续任务跳过


class CustomPythonOperator(PythonOperator):
    template_fields = {'skip_condition', 'skip_message'} | set(PythonOperator.template_fields)

    @apply_defaults
    def __init__(self, skip_condition, skip_message, **kwargs) -> None:
        self.skip_condition = skip_condition
        self.skip_message = skip_message
        super().__init__(**kwargs)

    def execute(self, context) -> Optional[str]:
        print("Check if the BulkAPI can be skipped")
        if self.skip_condition == "True":
            print("start custom PythonOperator")
            return super().execute(context)
        else:
            raise AirflowSkipException(self.skip_message)


# 启动GKE工作负载的任务定义
start_update_operator = CustomPythonOperator(
task_id=task_name,
provide_context=True,
python_callable=start_and_update,
op_args=[DATASET, TABLE, USE_CASE, use_case_data],
skip_condition="{{ ti.xcom_pull(task_ids=['prevState_'+ti.task_id])[0] }}",
skip_message="{{ ti.xcom_pull(task_ids=['prevState_'+ti.task_id], key='skip_message')[0] }}",
# trigger_rule="all_done",
dag=dag)

错误原因 #

Airflow在2.3及以上版本中对Jinja模板内的XCom获取逻辑做了性能优化：

• 在v2.2.5及更早版本中，ti.xcom_pull()即使只拉取单个任务的XCom，返回的也是包含单个元素的列表，因此需要通过[0]索引取值。
• 升级到v2.5.1后，模板内调用ti.xcom_pull()会返回LazyXComAccess对象而非直接的列表。这个对象是惰性加载的，只有在实际需要取值时才会查询数据库拉取XCom数据；当拉取的是单个任务的XCom时，该对象会自动解析为对应的值，不再以列表形式返回，因此使用[0]索引会触发"无元素0"的错误。

关键变更说明 #

Airflow引入LazyXComAccess的核心目的是减少不必要的数据库查询，提升模板渲染性能。该对象实现了自动解析逻辑：

• 当拉取多个任务的XCom时，它会在取值时返回列表；
• 当拉取单个任务的XCom时，直接返回对应的值，无需手动处理索引。

在你的代码中，task_ids=['prevState_'+ti.task_id]本质是单个任务ID，因此xcom_pull返回的LazyXComAccess对象直接对应目标XCom值，不需要再用[0]索引提取。

解决方案 #

直接移除模板字符串中的[0]索引即可，修改后的模板如下：

skip_condition="{{ ti.xcom_pull(task_ids=['prevState_'+ti.task_id]) }}"
skip_message="{{ ti.xcom_pull(task_ids=['prevState_'+ti.task_id], key='skip_message') }}"

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