如何从XCom拉取字典对象并传递给DataprocSubmitJobOperator

问题根源 #

核心问题在于提前将XCom拉取的模板字符串赋值给了job_args变量，这会让job_args在DAG解析阶段就被固定为字符串类型，即便开启了render_template_as_native_obj=True，也无法让Airflow在任务执行阶段将其渲染为字典。此外，PythonOperator的参数传递方式也存在同样的模板字符串误用问题。

解决方案步骤 #

1. 保留render_template_as_native_obj=True配置：这个参数是关键，它会让Airflow在渲染模板时返回原生Python对象（如字典、列表），而非JSON序列化后的字符串。
2. 修正PythonOperator的参数传递：不要用模板字符串传递message，而是在Python可调用函数内部通过任务实例直接拉取Pub/Sub Sensor的XCom数据。
3. 直接在DataprocSubmitJobOperator的job参数中使用模板语法：跳过提前赋值job_args的步骤，直接在submit_job字典的spark_job字段中嵌入XCom拉取的模板，让Airflow在任务执行时自动渲染为字典。

修正后的完整代码 #

dag = DAG(
    dag_id=dag_id,
    schedule_interval=None,
    default_args=default_args,
    render_template_as_native_obj=True  # 保留该配置，确保返回原生对象
)

with dag:        
    t1 = PubSubPullSensor(
        task_id='pull-messages',
        project="projectname",
        ack_messages=True,
        max_messages=1,
        subscription="subscribtionname"
    )

    def create_args_from_event(**context):
        # 从context中获取任务实例，拉取t1的XCom数据
        message = context['ti'].xcom_pull(task_ids='pull-messages')
        # 这里处理message生成预期的job_args字典
        job_args = {
            "gcs_job": {
                "args": ["--foo=bar", "--foo2=bar2"],
                "jar_file_uris": ["gs://...."],
                "main_class": "com.xyz.something"
            }
        }
        # 将job_args存入XCom，key为'define_args'
        context['ti'].xcom_push(key='define_args', value=job_args)

    t2 = PythonOperator(
        task_id='define_args',
        python_callable=create_args_from_event,
        provide_context=True,  # 传递context给函数
    )

    # 直接在submit_job中使用模板语法拉取XCom的字典
    submit_job = {
        "reference": {"project_id": v_project_id},
        "placement": {"cluster_name": v_cluster_name},
        "spark_job": "{{ task_instance.xcom_pull(task_ids='define_args', key='define_args')['gcs_job'] }}"
    }
    
    spark_job_submit = DataprocSubmitJobOperator(
        task_id="XXXX",
        job=submit_job,
        location="us-central1",
        gcp_conn_id=v_conn_id,
        project_id=v_project_id
    )

    # 设置任务依赖
    t1 >> t2 >> spark_job_submit

关键说明 #

• render_template_as_native_obj=True的作用：当Airflow渲染{{ task_instance.xcom_pull(...) }}时，会直接返回存储在XCom中的字典对象，而非将其序列化为JSON字符串。
• 避免模板字符串提前赋值：如果将job_args赋值为"{{ ... }}"，它会被当作普通字符串处理，Airflow无法对其进行模板渲染。必须将模板语法直接放在需要接收字典的参数字段中（即submit_job['spark_job']）。
• Python函数内获取XCom：通过context['ti'].xcom_pull()获取上游任务的XCom数据，比在op_kwargs中传递模板字符串更可靠，也避免了字符串转义问题。

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