如何在另一个达格气流中创建达格斯

我正在尝试拥有一个主匕首，它将根据我的需要创建更多的匕首。我在气流.cfg dags_folder中有以下python文件。此代码在数据库中创建主 dag。此主 dag 应读取文本文件，并应为文本文件中的每一行创建 dag。但是，在主 dag 中创建的 dag 不会添加到数据库中。创建它的正确方法是什么？版本详细信息：蟒蛇版本：3.7阿帕奇气流版本：1.10.8import datetime as dtfrom airflow import DAGfrom airflow.operators.bash_operator import BashOperatorfrom airflow.operators.python_operator import PythonOperatorroot_dir = "/home/user/TestSpace/airflow_check/res"print("\n\n ===> \n Dag generator")default_args = { 'owner': 'airflow', 'start_date': dt.datetime(2020, 3, 22, 00, 00, 00), 'concurrency': 1, 'retries': 0}def greet(_name): message = "Greetings {} at UTC: {} Local: {}\n".format(_name, dt.datetime.utcnow(), dt.datetime.now()) f = open("{}/greetings.txt".format(root_dir), "a+") print("\n\n =====> {}\n\n".format(message)) f.write(message) f.close()def create_dag(dag_name): with DAG(dag_name, default_args=default_args, schedule_interval='*/2 * * * *', catchup=False ) as i_dag: i_opr_greet = PythonOperator(task_id='greet', python_callable=greet, op_args=["{}_{}".format("greet", dag_name)]) i_echo_op = BashOperator(task_id='echo', bash_command='echo `date`') i_opr_greet >> i_echo_op return i_dagdef create_all_dags(): all_lines = [] f = open("{}/../dag_names.txt".format(root_dir), "r") for x in f: all_lines.append(str(x)) f.close() for line in all_lines: print("Dag creation for {}".format(line)) globals()[line] = create_dag(line)with DAG('master_dag', default_args=default_args, schedule_interval='*/1 * * * *', catchup=False ) as dag: echo_op = BashOperator(task_id='echo', bash_command='echo `date`') create_op = PythonOperator(task_id='create_dag', python_callable=create_all_dags) echo_op >> create_op

查看完整描述

2 回答