Airflow 任务能否在运行时动态生成 DAG？

Question

我有一个不规则上传的上传文件夹。对于每个上传的文件，我想生成一个特定于该文件的 DAG。

我的第一个想法是使用 FileSensor 来执行此操作，该文件传感器监视上传文件夹，并以新文件的存在为条件，触发创建单独 DAG 的任务。从概念上讲：

Sensor_DAG (FileSensor -> CreateDAGTask)

|-> File1_DAG (Task1 -> Task2 -> ...)
|-> File2_DAG (Task1 -> Task2 -> ...)

在我最初的实现中，CreateDAGTask是PythonOperator通过将它们放置在全局命名空间中来创建 DAG 全局变量（请参阅此 SO 答案），如下所示：

Sensor_DAG (FileSensor -> CreateDAGTask)

|-> File1_DAG (Task1 -> Task2 -> ...)
|-> File2_DAG (Task1 -> Task2 -> ...)

主 DAG 然后通过一个调用这个逻辑PythonOperator：

# File-sensing DAG
default_args = {
    "depends_on_past" : False,
    "start_date"      : datetime(2020, 7, 16),
    "retries"         : 1,
    "retry_delay"     : timedelta(hours=5),
}
with DAG("Sensor_DAG", default_args=default_args,
         schedule_interval= "50 * * * *", catchup=False, ) as sensor_dag:

    start_task  = DummyOperator(task_id="start")
    stop_task   = DummyOperator(task_id="stop")
    sensor_task = FileSensor(task_id="my_file_sensor_task",
                             poke_interval=60,
                             filepath=UPLOAD_LOCATION)
    process_creator_task = PythonOperator(
        task_id="process_creator",
        python_callable=generate_dags_for_files,
    )
    start_task >> sensor_task >> process_creator_task >> stop_task

但这不起作用，因为到process_creator_task运行时，Airflow 已经解析了全局变量。解析时间后的新全局变量无关紧要。

临时解决方案

每个Airflow 动态 DAG 和任务 Ids，我可以通过FileSensor完全省略任务并让 Airflow 在每个调度程序心跳时生成每个文件的任务来实现我想要做的事情，用只执行来替换 Sensor_DAG generate_dags_for_files：更新：没关系-虽然这确实在仪表板中创建了一个 DAG，但实际执行会遇到“DAG 似乎丢失”的问题：

from airflow import DAG
from airflow.operators.dummy_operator import DummyOperator
from airflow.operators.python_operator import PythonOperator
from airflow.contrib.sensors.file_sensor import FileSensor
from datetime import datetime, timedelta
from pathlib import Path

UPLOAD_LOCATION = "/opt/files/uploaded"

# Dynamic DAG generation task code, for the Sensor_DAG below
def generate_dags_for_files(location=UPLOAD_LOCATION, **kwargs):
    dags = []
    for filepath in Path(location).glob('*'):
        dag_name = f"process_{filepath.name}"
        dag = DAG(dag_name, schedule_interval="@once", default_args={
            "depends_on_past": True,
            "start_date": datetime(2020, 7, 15),
            "retries": 1,
            "retry_delay": timedelta(hours=12)
        }, catchup=False)
        dag_task = DummyOperator(dag=dag, task_id=f"start_{dag_name}")

        dags.append(dag)

        # Try to place the DAG into globals(), which doesn't work
        globals()[dag_name] = dag

    return dags

这确实意味着我无法再使用poke_interval参数来调节文件夹轮询的频率FileSensor；相反，Airflow 每次收集 DAG 时都会轮询文件夹。

这是这里最好的模式吗？

其他相关的 StackOverflow 线程

• 为每个文件和Airflow运行 Airflow DAG：为每个文件运行 DAG 的正确方法：相同的用例，但接受的答案使用两个静态 DAG，大概具有不同的参数。
• 在 Airflow 中创建动态工作流的正确方法- 接受的答案通过复杂的 XCom 设置动态创建任务，而不是 DAG。

Answer 1

简而言之：如果任务写入DagBag读取的位置，是的，但最好避免需要这样做的模式。您想要在任务中自定义创建的任何 DAG 可能应该是静态、高度参数化、条件触发的 DAG。 y2k-shubham 提供了这种设置的一个很好的例子，我很感谢他在这个问题的评论中提供的指导。

也就是说，以下是可以完成问题所要求的方法，无论它是多么糟糕的想法，在日益笨手笨脚的情况下：

• 如果您从变量动态生成 DAG（如下所示），请修改变量。
• 如果您从配置文件列表动态生成 DAG，请将新的配置文件添加到您从中提取配置文件的位置，以便在下一个 DAG 集合上生成新的 DAG。
• 使用 Jinja 模板之类的东西在文件夹中写入一个新的 Python 文件dags/。

要在任务运行后保留对任务的访问权限，您必须保持新的 DAG 定义稳定并在未来的仪表板更新/DagBag集合中可访问。否则，Airflow 仪表板将无法呈现太多相关内容。