25

我正在尝试使用 Airflow 来执行一个简单的任务 python。

from __future__ import print_function
from airflow.operators.python_operator import PythonOperator
from airflow.models import DAG
from datetime import datetime, timedelta


from pprint import pprint

seven_days_ago = datetime.combine(datetime.today() - timedelta(7),
                                  datetime.min.time())

args = {
    'owner': 'airflow',
    'start_date': seven_days_ago,
}

dag = DAG(dag_id='python_test', default_args=args)


def print_context(ds, **kwargs):
    pprint(kwargs)
    print(ds)
    return 'Whatever you return gets printed in the logs'

run_this = PythonOperator(
    task_id='print',
    provide_context=True,
    python_callable=print_context,
    dag=dag)

如果我尝试，例如：

气流测试 python_test 打印 2015-01-01

有用！

现在我想把我的def print_context(ds, **kwargs)函数放在其他 python 文件中。所以我创建了一个名为：simple_test.py 的文件并更改：

run_this = PythonOperator(
    task_id='print',
    provide_context=True,
    python_callable=simple_test.print_context,
    dag=dag)

现在我尝试再次运行：

气流测试 python_test 打印 2015-01-01

好！它仍然有效！

但是如果我创建一个模块，例如，带有文件的工作模块SimplePython.py，导入（from worker import SimplePython）它并尝试：

气流测试 python_test 打印 2015-01-01

它给出了信息：

ImportError：没有名为worker的模块

问题：

  1. 是否可以在 DAG 定义中导入模块？
  2. Airflow+Celery 将如何在工作节点之间分发所有必要的 python 源文件？
4

4 回答 4

14

您可以按照以下方式打包 DAG 的依赖项：

https://airflow.apache.org/concepts.html#packaged-dags

为此，您可以在 zip 文件的根目录中创建一个包含 dag 的 zip 文件，并将额外的模块解压缩到目录中。例如，您可以创建一个如下所示的 zip 文件：

my_dag1.py
my_dag2.py
package1/__init__.py
package1/functions.py

Airflow 将扫描 zip 文件并尝试加载 my_dag1.py 和 my_dag2.py。它不会进入子目录，因为这些被认为是潜在的包。

使用 CeleryExecutor 时，您需要手动同步 DAG 目录，Airflow 不会为您处理这些：

https://airflow.apache.org/configuration.html?highlight=scaling%20out%20celery#scaling-out-with-celery

worker 需要访问其 DAGS_FOLDER，并且您需要通过自己的方式同步文件系统

于 2016-09-28T13:18:59.023 回答
11

虽然按照文档中的说明将 dag 打包成 zip 是我见过的唯一受支持的解决方案，但您也可以导入 dags 文件夹中的模块。如果您使用 puppet 和 git 等其他工具自动同步 dags 文件夹，这将非常有用。

我从问题中不清楚您的目录结构，所以这里是一个基于典型 python 项目结构的示例 dags 文件夹：

└── airflow/dags  # root airflow dags folder where all dags live
    └── my_dags  # git repo project root
        ├── my_dags  # python src root (usually named same as project)
        │   ├── my_test_globals.py  # file I want to import
        │   ├── dag_in_package.py 
        │   └── dags 
        │        └── dag_in_subpackage.py
        ├── README.md  # also setup.py, LICENSE, etc here
        └── dag_in_project_root.py

我遗漏了（必需的 [ 1 ]）__init__.py文件。注意三个示例 dag 的位置。您几乎可以肯定只会将这些地方之一用于您的所有 dag。为了举例，我将它们都包括在这里，因为它对导入无关紧要。my_test_globals从其中任何一个导入：

from my_dags.my_dags import my_test_globals

我相信这意味着气流运行时将 python 路径设置为 dags 目录，因此 dags 文件夹的每个子目录都可以视为 python 包。在我的例子中，额外的中间项目根目录妨碍了典型的包内绝对导入。因此，我们可以像这样重组这个气流项目：

└── airflow/dags  # root airflow dags folder where all dags live
    └── my_dags  # git repo project root & python src root
        ├── my_test_globals.py  # file I want to import
        ├── dag_in_package.py 
        ├── dags 
        │    └── dag_in_subpackage.py
        ├── README.md  # also setup.py, LICENSE, etc here
        └── dag_in_project_root.py

这样导入看起来就像我们期望的那样：

from my_dags import my_test_globals
于 2017-11-22T16:28:48.687 回答
2

对于您的第二个问题：Airflow+Celery 将如何在工作节点之间分发所有必要的 python 源文件？

来自文档：工作人员需要有权访问其 DAGS_FOLDER，并且您需要按自己的方式同步文件系统。一个常见的设置是将您的 DAGS_FOLDER 存储在 Git 存储库中，并使用 Chef、Puppet、Ansible 或您用于在环境中配置机器的任何工具在机器之间同步它。如果你所有的盒子都有一个共同的挂载点，那么在那里共享你的管道文件应该也可以

http://pythonhosted.org/airflow/installation.html?highlight=chef

于 2016-03-31T07:42:30.633 回答
1

对于您的第一个问题，这是可能的。

__init__.py而且我想您应该在同一目录下创建一个名为的空文件SimplePython.py（在您的情况下是worker目录）。通过这样做，该worker目录将被视为一个python模块。

然后在您的 DAG 定义中，尝试from worker.SimplePython import print_context.

在您的情况下，我想如果您为气流编写一个插件会更好，因为您可能希望在不删除自定义功能的情况下升级气流核心项目。

于 2015-11-24T21:16:06.140 回答