Airflow中如何将XCom变量传入EventBridgePutEventsOperator？

你遇到的错误是因为Jinja模板渲染后，task_instance.xcom_pull返回的列表会被转换成字符串类型，而EventBridgePutEventsOperator的entries参数要求传入list或tuple类型，导致参数校验失败。

方法1：使用XComArg（Airflow 2.0+推荐） #

Airflow 2.0及以上版本支持用XComArg直接传递Python对象，无需通过Jinja模板，能完整保留原数据类型：

from airflow.operators.python import PythonOperator
from airflow.providers.amazon.aws.operators.event_bridge import EventBridgePutEventsOperator
from airflow.models.xcom_arg import XComArg

def extract_and_format(ti):
    events = ti.xcom_pull(task_ids=f"fetch_historical_events")
    formatted_events = []

    for event_payload in events:
        formatted_payload = {
            "Source": "some source",
            "DetailType": "some detail type",
            "EventBusName": "some bus name",
            "Detail": event_payload,
        }
        formatted_events.append(formatted_payload)

    return formatted_events

# 定义格式化任务
extract_and_format_task = PythonOperator(
    task_id="extract_and_format",
    python_callable=extract_and_format,
    provide_context=True,
)

# 用XComArg直接传递输出给EventBridge算子
put_event_task = EventBridgePutEventsOperator(
    task_id="put_event_task",
    entries=XComArg(extract_and_format_task),
)

# 设置任务依赖
extract_and_format_task >> put_event_task

方法2：合并任务逻辑（兼容Airflow 1.x） #

如果使用Airflow 1.x版本，或者不想用XComArg，可以直接在PythonOperator中完成事件格式化+推送的逻辑：

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

def extract_and_format_put_events(ti):
    events = ti.xcom_pull(task_ids=f"fetch_historical_events")
    formatted_events = []

    for event_payload in events:
        formatted_payload = {
            "Source": "some source",
            "DetailType": "some detail type",
            "EventBusName": "some bus name",
            "Detail": json.dumps(event_payload),  # 注意：Detail需要是JSON字符串
        }
        formatted_events.append(formatted_payload)
    
    # 直接调用EventBridge API推送事件
    client = boto3.client('events')
    response = client.put_events(Entries=formatted_events)
    return response

put_event_task = PythonOperator(
    task_id="put_event_task",
    python_callable=extract_and_format_put_events,
    provide_context=True,
)

关键注意点 #

• 禁止用Jinja模板传递复杂Python对象：Jinja会自动将对象序列化为字符串，无法直接还原为原类型（强行用ast.literal_eval存在安全风险，不推荐）。
• Detail字段必须是JSON字符串：EventBridge要求该字段为JSON格式的字符串，若原event_payload是Python字典，需用json.dumps()转换。

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