使用外部文件在气流中动态创建任务答案

【问题标题】：Create tasks dynamically in airflow with external file使用外部文件在气流中动态创建任务
【发布时间】：2020-04-28 06:36:47
【问题描述】：

我正在尝试创建一个 DAG，它根据位于存储中的 JSON 文件动态生成任务。我一步一步地遵循了这个指南：

https://bigdata-etl.com/apache-airflow-create-dynamic-dag/

但是 DAG 卡住了以下消息：

是否可以读取外部文件并使用它在 Composer 中动态创建任务？当我只从气流变量中读取数据时，我可以这样做，但是当我读取外部文件时，dag 卡在isn't available in the web server's DagBag object 状态。我需要从外部文件中读取，因为 JSON 的内容会随着每次执行而改变。

我正在使用composer-1.8.2-airflow-1.10.2。

我读到了一个类似问题的答案：

Dynamic task definition in Airflow

但我并没有尝试基于单独的任务创建任务，仅基于外部文件。

这是我的第二种方法，也陷入了错误状态：

import datetime

import airflow
from airflow.operators import bash_operator
from airflow.operators.dummy_operator import DummyOperator
from airflow.models import Variable
import json
import os

products = json.loads(Variable.get("products"))

default_args = {
    'owner': 'Composer Example',
    'depends_on_past': False,
    'email': [''],
    'email_on_failure': False,
    'email_on_retry': False,
    'retries': 0,
    'retry_delay': datetime.timedelta(minutes=5),
    'start_date': datetime.datetime(2020, 1, 10),
}

with airflow.DAG(
        'json_test2',
        default_args=default_args,
        # Not scheduled, trigger only
        schedule_interval=None) as dag:


        # Print the dag_run's configuration, which includes information about the
        # Cloud Storage object change.
        def read_json_file(file_path):
            if os.path.exists(file_path):
                with open(file_path, 'r') as f:
                    return json.load(f)

        def get_run_list(files):
            run_list = []
            #The file is uploaded in the storage bucket used as a volume by Composer
            last_exec_json = read_json_file("/home/airflow/gcs/data/last_execution.json")
            date = last_exec_json["date"]
            hour = last_exec_json["hour"]
            for file in files:
                #Testing by adding just date and hour
                name = file['name']+f'_{date}_{hour}'
                run_list.append(name)
            return run_list

        rl = get_run_list(products)

        start = DummyOperator(task_id='start', dag=dag)
        end = DummyOperator(task_id='end', dag=dag)

        for name in rl:
            tsk = DummyOperator(task_id=name, dag=dag)
            start >> tsk >> end

【问题讨论】：

检查您的网络服务器日志，很可能将您的 DAG 导入 DAGag 时出错
您的回答描述了我遵循的完全相同的步骤，但它不起作用。我稍后会重试并更新您。

标签： python airflow google-cloud-composer orchestration

【解决方案1】：

可以创建基于位于 Cloud Storage 存储分区中的 JSON 文件动态生成任务的 DAG。我遵循了您提供的指南，它在我的情况下非常有效。

首先你需要将你的JSON配置文件上传到$AIRFLOW_HOME/dags目录，然后将DAG python文件上传到相同的路径（你可以在位于bucket中的airflow.cfg文件中找到路径）。

稍后，您将能够在 Airflow UI 中看到 DAG：

您可以看到日志DAG isn't available in the web server's DagBag object，DAG 在 Airflow Web Server 上不可用。但是，可以将 DAG 安排为活动状态，因为 Airflow Scheduler 与 Airflow Web Server 独立工作。

当大量 DAG 一次加载到 Composer 环境中时，它可能会使环境过载。由于 Airflow 网络服务器位于 Google 管理的项目中，因此只有某些类型的更新会导致网络服务器容器重新启动，例如添加或升级其中一个 PyPI 包或更改 Airflow 设置。解决方法是添加一个虚拟环境变量：

在 GCP 中打开 Composer 实例
ENVIRONMENT VARIABLE标签
Edit，然后添加环境变量和Submit

您可以使用以下命令重新启动它：

gcloud composer environments update ${ENVIRONMENT_NAME}  --location=${ENV_LOCATION}  --update-airflow-configs=core-dummy=true
gcloud composer environments update ${ENVIRONMENT_NAME}  --location=${ENV_LOCATION}  --remove-airflow-configs=core-dummy

希望以上信息对您有用。

【讨论】：