损坏的 DAG：[/home/airflow/gcs/dags/airflow_test_task.py] 名称“cfg”未定义

Question

我是 python 和气流的新手，我正在使用 GCP 作曲家环境来创建 DAG。
在这个 python 代码中，我创建了两个任务，一个是读取 zip 或 csv 文件，另一个是创建 dataproc 集群。在一项任务中，我正在调用一种方法 readYML，它正在读取 yml 配置文件以获取 dataproc 集群参数（如 cluster-name、project_id 等）以及我在第二项任务中进一步使用的相同参数，请参阅下面的代码以更好地理解

# Importing Modules

from airflow import DAG

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

from airflow.models import Variable
import yaml
from google.cloud import storage
from airflow.contrib.operators import dataproc_operator
import pandas as pd


global cfg

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


def iterate_bucket():
    global blobs
    bucket_name = 'europe-west1-airflow-test-9bbb5fc7-bucket'
    storage_client = storage.Client.from_service_account_json(
        '/home/airflow/gcs/data/service_account_key_gcp_compute_bmg.json')
    bucket = storage_client.get_bucket(bucket_name)
    blobs = bucket.list_blobs()


def print_PcsvData():
    iterate_bucket()
    readYML()
    global readPcsv

    for blob in blobs:
        if "physical.zip" in blob.name:
            print("hello : ", blob.name)
            file_name = "/home/airflow/gcs/" + blob.name

    with ZipFile(file_name, 'r') as zip:
        # printing all the contents of the zip file
        for info in zip.infolist():
            readfilename = info.filename
            print(readfilename)

    readPcsv = pd.read_csv("/home/airflow/gcs/data/" + readfilename)

    print("physi cal.csv : ", readPcsv)
    print('Done!')


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   

t1 = PythonOperator(task_id='Raw1', python_callable=print_PcsvData, 
dag=dag)

create_dataproc_cluster = dataproc_operator.DataprocClusterCreateOperator(
    task_id='create_dataproc_cluster',
    project_id=cfg['configs']['project_id'],
    cluster_name=cfg['configs']['cluster_name'],
    num_workers=cfg['configs']['num_workers'],
    zone=cfg['configs']['zone'],
    master_machine_type=cfg['configs']['master_machine_type'],
    worker_machine_type=cfg['configs']['worker_machine_type'],
    dag=dag)

t1 >> create_dataproc_cluster

在这段代码中，我想全局使用 cfg 变量，在默认 args 中我也想访问这个变量但我收到错误，我不知道它的范围相关问题，甚至我在 readYML 方法中声明了 cfg 变量也但仍然存在错误。 任何帮助，将不胜感激。 提前致谢

Answer 1

检查下面您应该使用的 DAG 文件：

你应该做的一些改变：

• 更改全局变量并从函数返回值
• 永远不要使用datetime.now() - https://airflow.apache.org/faq.html#what-s-the-deal-with-start-date

更新文件：

# Importing Modules

from airflow import DAG

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

from airflow.models import Variable
import yaml
from google.cloud import storage
from airflow.contrib.operators import dataproc_operator
import pandas as pd



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


def iterate_bucket():
    bucket_name = 'europe-west1-airflow-test-9bbb5fc7-bucket'
    storage_client = storage.Client.from_service_account_json(
        '/home/airflow/gcs/data/service_account_key_gcp_compute_bmg.json')
    bucket = storage_client.get_bucket(bucket_name)
    blobs = bucket.list_blobs()
    return blobs


def print_PcsvData():
    blobs = iterate_bucket()

    for blob in blobs:
        if "physical.zip" in blob.name:
            print("hello : ", blob.name)
            file_name = "/home/airflow/gcs/" + blob.name

    with ZipFile(file_name, 'r') as zip:
        # printing all the contents of the zip file
        for info in zip.infolist():
            readfilename = info.filename
            print(readfilename)

    readPcsv = pd.read_csv("/home/airflow/gcs/data/" + readfilename)

    print("physi cal.csv : ", readPcsv)
    print('Done!')
    return readPcsv

dag_name = Variable.get("dag_name")

cfg = readYML()

default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'start_date': airflow.utils.dates.days_ago(2),
    '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   

t1 = PythonOperator(task_id='Raw1', python_callable=print_PcsvData, 
dag=dag)

create_dataproc_cluster = dataproc_operator.DataprocClusterCreateOperator(
    task_id='create_dataproc_cluster',
    project_id=cfg['configs']['project_id'],
    cluster_name=cfg['configs']['cluster_name'],
    num_workers=cfg['configs']['num_workers'],
    zone=cfg['configs']['zone'],
    master_machine_type=cfg['configs']['master_machine_type'],
    worker_machine_type=cfg['configs']['worker_machine_type'],
    dag=dag)

t1 >> create_dataproc_cluster