Airflow中如何将XCom的值作为Operator的入参使用

问题根因 #

你遇到的报错和疑问可以从以下几个点定位：

• 模板语法错误：Airflow使用Jinja2模板渲染动态参数，要求模板表达式必须包裹在两层大括号{{ }}中，你仅使用了一层大括号，导致系统直接把你写的表达式字符串当做连接ID去查询，自然找不到对应配置
• XCom取值逻辑错误：task1返回的XCom结构是[{'SOURCE_HOST': 'TEST_HOST'}]，直接调用xcom_pull拿到的是完整数组，需要嵌套取值才能拿到最终可用的连接ID
• 参数模板支持缺失：需要确认你使用的MsSqlToOracleTransfer的mssql_conn_id参数是否在Operator的template_fields列表中，不在列表内的参数不会被模板渲染
• 任务依赖缺失：你没有声明task1和task2的执行顺序，可能出现task2先执行，拉取不到XCom的问题
• 冗余代码：print_xcom函数没有被注册为任务也没有被调用，完全可以删除，不需要额外函数中转就可以直接在Operator参数中拉取XCom

解决步骤 #

1. 修正模板语法和取值逻辑 #

修改load_from_db任务的mssql_conn_id参数：

load_from_db = MsSqlToOracleTransfer(
    task_id='task2',
    # 两层大括号包裹模板表达式，嵌套取XCom中的SOURCE_HOST值
    mssql_conn_id="{{ ti.xcom_pull(task_ids='task1')[0]['SOURCE_HOST'] }}",
    oracle_conn_id='conn_def_orc',
    sql='test.sql',
    oracle_table="oracle_table"
)

其中ti是task_instance的内置别名，可以直接使用。

2. 解决参数不支持模板的问题 #

如果修改语法后仍未生效，说明mssql_conn_id不在当前Operator的模板字段列表中，可通过自定义Operator扩展模板字段解决：

# 继承原有Operator，新增mssql_conn_id到模板字段
class CustomMsSqlToOracleTransfer(MsSqlToOracleTransfer):
    template_fields = MsSqlToOracleTransfer.template_fields + ('mssql_conn_id',)

# 使用自定义Operator创建任务
load_from_db = CustomMsSqlToOracleTransfer(
    # 其余参数不变
)

3. 添加任务依赖 #

在DAG末尾添加执行顺序声明，确保task1执行完成后再运行task2：

test_l >> load_from_db

4. 清理冗余代码 #

删除未使用的print_xcom函数，以及无意义的tasks.append(load_from_db)语句即可。

修正后核心代码示例 #

from datetime import timedelta
from airflow import DAG
# 导入你需要的其他Operator

# 自定义Operator（如果需要的话）
class CustomMsSqlToOracleTransfer(MsSqlToOracleTransfer):
    template_fields = MsSqlToOracleTransfer.template_fields + ('mssql_conn_id',)

args = {
    # 你的默认参数
}
tmpl_search_path = "/your/template/path"

with DAG(
        schedule_interval='@daily',
        dagrun_timeout=timedelta(minutes=120),
        default_args=args,
        template_searchpath=tmpl_search_path,
        catchup=False,
        dag_id='test'
) as dag:
    
    test_l = OracleLoadOperator(
            task_id = "task1",
            oracle_conn_id="orcl_conn_id",
            object_name='table'
    )

    load_from_db = CustomMsSqlToOracleTransfer(
                task_id= 'task2',
                mssql_conn_id = "{{ ti.xcom_pull(task_ids='task1')[0]['SOURCE_HOST'] }}",
                oracle_conn_id = 'conn_def_orc',
                sql= 'test.sql',
                oracle_table = "oracle_table"
    )

    test_l >> load_from_db

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