如何阅读动态参数气流运算符？

Question

我是 python 和气流 dag 的新手。 我正在关注答案部分中提到的以下链接和代码。
How to pass dynamic arguments Airflow operator?

我在读取 yaml 文件时遇到问题，在 yaml 文件中我有一些与配置相关的参数。

configs:
    cluster_name: "test-cluster"
    project_id: "t***********"
    zone: "europe-west1-c"
    num_workers: 2
    worker_machine_type: "n1-standard-1"
    master_machine_type: "n1-standard-1"

在 DAG 脚本中，我创建了一个将创建集群的任务，在执行此任务之前，我们需要传递给它的所有参数 default_args 参数，如集群名称、项目 ID 等。为了读取我创建的那些参数一种readYML方法。见下面的代码

from airflow import DAG
from airflow.operators.python_operator import PythonOperator
from datetime import datetime, timedelta
from zipfile import ZipFile
from airflow.contrib.operators import dataproc_operator

from airflow.models import Variable
import yaml

def readYML():
     print("inside readYML")
     global cfg
     file_name = "/home/airflow/gcs/data/cluster_config.yml"
     with open(file_name, 'r') as ymlfile:
          cfg = yaml.load(ymlfile)
     print(cfg['configs']['cluster_name'])

 # Default Arguments
 readYML()

 dag_name = Variable.get("dag_name")

  default_args = {
     'owner': 'airflow',
     'depends_on_past': False,
     'start_date': datetime.now(),
     'email': ['airflow@example.com'],
     'email_on_failure': False,
     'email_on_retry': False,
     'retries': 1,
     'retry_delay': timedelta(minutes=5),
     #'cluster_name': cfg['configs']['cluster_name'],    
    }

    # Instantiate a DAG

    dag = DAG(dag_id='read_yml', default_args=default_args, 
    schedule_interval=timedelta(days=1))

    # Creating Tasks
    Task1 = DataprocClusterCreateOperator(
    task_id='create_cluster',
    dag=dag
    )

在这段代码中没有错误，当我在 GCP 作曲家环境中上传时，没有显示错误通知，但是这个 DAG 不可运行，没有运行按钮。

请参阅随附的屏幕截图。 我正在使用 python 3 和气流 composer-1.7.2-airflow-1.10.2 版本。

Answer 1

根据 Cloud Composer 文档中的 Data Stored in Cloud Storage 页面：

为避免网络服务器错误，请确保网络服务器解析 DAG（未运行）所需的数据在 dags/ 文件夹中可用。否则，网络服务器无法访问数据或加载 Airflow 网络界面。

您的 DAG 正在尝试打开 /home/airflow/gcs/data 下的 YAML 文件，该文件在网络服务器上不存在。将文件放在您的 GCS 存储桶中的 dags/ 文件夹下，调度程序、工作人员和 Web 服务器将可以访问该文件，并且 DAG 将在 Web UI 中运行。