如何通过单个脚本生成多个气流dag？

Question

我想用一个脚本生成多个气流dags。 dag名称应为“test_parameter”。以下是我的脚本：

from datetime import datetime

# Importing Airflow modules
from airflow.models import DAG
from airflow.operators import DummyOperator

# Specifying the default arguments for the DAG
default_args = {
    'owner': 'Test',
    'start_date': datetime.now()
    }

parameter_list = ["abc", "pqr", "xyz"]

for parameter in parameter_list:
    dag = DAG("test_"+parameter,
              default_args=default_args,
              schedule_interval=None)
    dag.doc_md = "This is a test dag"

    # Creating Start Dummy Operator
    start = DummyOperator(
        task_id="start",
        dag=dag)

    # Creating End Dummy Operator
    end = DummyOperator(
        task_id="end",
        dag=dag)

    # Design workflow of tasks in the dag
    end.set_upstream(start)

所以在这种情况下，它应该创建3天：“test abc”，“test pqr”和“test xyz”。

但是在运行脚本时，它只创建了一个dag“test_xyz”。有关如何解决此问题的任何见解。提前致谢 ：）

Answer 1

我想问题是dag对象'start'和'end'被forloop覆盖，因此只保留最后一个值。

奇怪的是，虽然你不能动态创建dag，但你可以通过循环动态创建任务。也许这有帮助。

for i in range(3):
    t1 = BashOperator(
    task_id='Success_test'+str(i),
    bash_command='cd home',
    dag=dag)

    slack_notification.set_upstream(t1)

Answer 2

您可以在全局命名空间中注册动态创建的dag。

例如：

global()[parameter] = dag

Answer 3

是的，您可以保存每个DAG的配置，即存储内部。例如，您可以将配置保存在永久存储（DB）中，然后获取配置并将结果保存在缓存中。这样做主要是因为我们希望每次刷新DAG脚本时都阻止dag脚本从DB中获取配置。因此，我们使用缓存并保存其到期时间。您可以参考此article了解如何创建动态DAG

for i in range(10):
  dag_id = 'foo_{}'.format(i)
  globals()[dag_id] = DAG(dag_id)

反过来，您还希望创建动态子DAG和动态任务。希望能帮助到你 ：-）