动态 DAG 创建在 Apache 气流中未按预期工作答案

【问题标题】：Dynamic DAG Creation Not working as expected in Apache airflow动态 DAG 创建在 Apache 气流中未按预期工作
【发布时间】：2019-06-12 20:41:04
【问题描述】：

我试图了解如何在 Apache 气流中创建动态 dag，因为我需要它来在我的项目中创建动态 dag。

以下是我关注的链接：Dynamic DAG creation in Apache airflow

以下是用于创建示例 hello world 动态 DAGS 的代码块。（基于输入参数的动态 DAG 创建）。

from datetime import datetime

from airflow import DAG

from airflow.operators.python_operator import PythonOperator


def create_dag(dag_id,
               schedule,
               dag_number,
               default_args):

    def hello_world_py(*args):
        print('Hello World')
        print('This is DAG: {}'.format(str(dag_number)))

    dag = DAG(dag_id,
              schedule_interval=schedule,
              default_args=default_args)

    with dag:
        t1 = PythonOperator(
            task_id='hello_world',
            python_callable=hello_world_py,
            dag_number=dag_number)

    return dag


# build a dag for each number in range(10)
for n in range(1, 10):
    dag_id = 'hello_world_{}'.format(str(n))

    default_args = {'owner': 'airflow',
                    'start_date': datetime(2018, 1, 1)
                    }

    schedule = '@daily'

    dag_number = n

    globals()[dag_id] = create_dag(dag_id,
                                  schedule,
                                  dag_number,
                                  default_args)

期望创建 9 个这样的 DAG。但是我可以看到，一旦我使用 python3 code_sample.py 编译上述代码块，它会创建 9 个 DAG，但是嵌入在 DAG 中的代码是完整的示例代码。

但据我了解，创建的 DAG 应该只有以下代码块，该代码块在上述示例代码块的 create_dag 方法中可用。

预期的 DAG 代码：

from datetime import datetime

 from airflow import DAG

 from airflow.operators.python_operator import PythonOperator
def hello_world_py(*args):
   print('Hello World')
   print('This is DAG: {}'.format(str(dag_number)))

dag = DAG(dag_id,
      schedule_interval=schedule,
      default_args=default_args)

with dag:
   t1 = PythonOperator(
    task_id='hello_world',
    python_callable=hello_world_py,
    dag_number=dag_number)

实际 DAG 代码：

from datetime import datetime

from airflow import DAG

from airflow.operators.python_operator import PythonOperator


def create_dag(dag_id,
               schedule,
               dag_number,
               default_args):

    def hello_world_py(*args):
        print('Hello World')
        print('This is DAG: {}'.format(str(dag_number)))

    dag = DAG(dag_id,
              schedule_interval=schedule,
              default_args=default_args)

    with dag:
        t1 = PythonOperator(
            task_id='hello_world',
            python_callable=hello_world_py,
            dag_number=dag_number)

    return dag


# build a dag for each number in range(10)
for n in range(1, 10):
    dag_id = 'hello_world_{}'.format(str(n))

    default_args = {'owner': 'airflow',
                    'start_date': datetime(2018, 1, 1)
                    }

    schedule = '@daily'

    dag_number = n

    globals()[dag_id] = create_dag(dag_id,
                                  schedule,
                                  dag_number,
                                  default_args)

让我知道是什么造成了上述问题

【问题讨论】：

这不是问题，但可以正常工作。昨天，我将我的多个 dag 移动到了动态 dag。每个 dag 的详细代码中显示的代码都是相同的。我使用 variable & xcome 让动态 dag 与不同的输入一起工作。

标签： python-3.x airflow

【解决方案1】：

点击“代码”选项卡时，您在 Airflow UI 中看到的代码就是整个 .py 文件源代码。看看这个函数是怎么实现的：

https://github.com/apache/airflow/blob/master/airflow/www/views.py#L437

【讨论】：