如何在Airflow DAG中将XCom变量作为参数传递给下游Dataproc任务？

报错原因 #

• 运行时变量直接在解析阶段调用：ti是Airflow任务运行时才会实例化的对象，你在定义Operator的阶段直接写ti.xcom_pull，DAG加载解析时找不到ti变量，自然会报未定义错误。
• 语法错误：现有代码中task_ids=simple_http未给上游任务ID加引号，且xcom_pull方法缺少右侧闭合括号。

解决方法 #

Airflow支持通过Jinja模板在运行时动态渲染参数，DataprocWorkflowTemplateInstantiateOperator的parameters字段默认支持模板解析，你直接用模板语法引用XCom值即可，修改后的代码如下：

dataproc_job = dataproc_operator.DataprocWorkflowTemplateInstantiateOperator(
    # The task id of your job
    task_id="dataproc_job",
    # The template id of your workflow
    template_id="newwf1",
    project_id='#######',
    region="us-central1",
    parameters={"TABLE_NAME": "{{ ti.xcom_pull(task_ids='simple_http') }}"}
)

注意事项 #

• 确保上游生成表名的任务ID和task_ids参数里的值完全一致，本示例中假设你的HTTP调用任务ID为simple_http，请根据实际情况修改。
• 确保上游任务开启了XCom推送：如果你用的是SimpleHttpOperator触发Cloud Function，需要给该任务设置do_xcom_push=True，才会把API返回的表名存入XCom。
• 如果Cloud Function返回的是JSON结构，你还可以在模板里直接解析取值，比如返回值里table_name字段存表名的话，写法为{{ ti.xcom_pull(task_ids='simple_http')['table_name'] }}

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