Python 如何使用传递的输入 json 配置的值来触发 dag 内的气流作业(job)

dag_run仅在活动DAG运行中可用(即，当您触发DAG并运行时)-指定dag_run.conf作为对象将不起作用，因为气流调度器将解析您的DAG(默认情况下每30秒一次)，并且不存在此类对象.

访问dag_run有两种方式:

1. 使用模板系统

CLUSTER_NAME = "my-cluster-for-{{ dag_run.conf['x'] }}{{ dag_run.conf['y'] }}{{ dag_run.conf['z'] }}"

仅当cluster_name字段是XyzCreateClusterOperator运算符中的模板字段时，此操作才有效.您可以通过查看文档字符串或操作员的气流documentation来判断字段是否已模板化.

如果XyzCreateClusterOperator是第三方Provider ，您可以查看Provider here的文档

您可以在模板here中看到所有可用项(dag_run是使用此模板语法可以访问的许多内容之一).

2. 您可以在python操作符中访问dag_run

import pendulum
from airflow.decorators import dag, task

@dag(
    schedule_interval=None,
    start_date=pendulum.datetime(2022, 4, 18),
    catchup=False,
)
def example_dag():
    @task()
    def dag_run_example_task(**context):
        context["dag_run"].conf.get("x")
        context["dag_run"].conf.get("y")
        context["dag_run"].conf.get("z")

如果cluster_name不是模板化字段，则可以使用Python操作符包装操作符并替换为:

import pendulum
from airflow.decorators import dag, task

@dag(
    schedule_interval=None,
    start_date=pendulum.datetime(2022, 4, 18),
    catchup=False,
)
def example_dag():
    @task()
    def dag_run_example_task(**context):
        cluster_name = "my-cluster-for-"+context["dag_run"].conf.get("x")+context["dag_run"].conf.get("y")+context["dag_run"].conf.get("z")

        create_cluster_spark = XyzCreateClusterOperator(
            task_id='create_cluster_spark',
            cluster_name=cluster_name,
            location=REGION,
            gcp_conn_id=CONN_ID,
        )

        create_cluster_spark.execute(context=context)

模板here中可用的相同对象在传递给PythonOperator的context对象中可用(还有一些其他对象).