如何在Kubernetes Pod Operator中正确传递XCom值作为参数

你的问题出在arguments参数的Jinja模板语法使用错误，导致XCom值无法被正确解析填充。

主要错误点 #

• 使用了嵌套大括号：{ti.xcom_pull(dag_id='{DAG_ID}', ...)}，Jinja模板中外层已经是模板标识，内部变量无需再加大括号包裹。
• 模板表达式未使用Airflow要求的{{ }}双大括号语法。

修正后的代码 #

get_report_task = kubernetes_pod_operator.KubernetesPodOperator(
    task_id="get_report_task",
    name="%s" % (DAG_ID.replace("_", "-")),
    namespace="dsm-kl",
    image=KUBER_IMAGE,
    image_pull_policy="Always",
    is_delete_operator_pod=True,
    # 正确使用Jinja模板语法引用XCom值
    arguments=["{{ ti.xcom_pull(dag_id=DAG_ID, task_ids='push_task', key='kube_values') }}"],
    labels={
        "vendor": "reversal",
        "job": "get_reports",
        "dag": DAG_ID,
        "task": "get_report_task",
        "task_id": "get_report_task",
    },
    retries=3,
    depends_on_past=False,
    dag=dag,
    resources={"request_memory": "512Mi", "limit_memory": "1024Mi"},
    affinity=DOC_AFFINITY,
)

额外注意事项 #

• 若kube_values是复杂数据结构（如字典、列表），需将其序列化为字符串后传递，可使用Jinja的tojson()过滤器：

  arguments=["{{ ti.xcom_pull(dag_id=DAG_ID, task_ids='push_task', key='kube_values') | tojson }}"],
  

• 确保push_task已成功推送XCom值，且DAG_ID变量在当前DAG上下文中可正常访问。

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