如果安装的模块在 zip 文件的顶层添加 .py 文件，Airflow 打包的 DAG 将不起作用答案

【问题标题】：Airflow packaged DAGs won't work if installed modules add .py file on top-level of the zip file如果安装的模块在 zip 文件的顶层添加 .py 文件，Airflow 打包的 DAG 将不起作用
【发布时间】：2021-07-08 12:56:28
【问题描述】：

我正在为一个数据科学家团队设置一个多用户气流集群，DAG 有多种用途（ETL、NLP、ML、NN...），其中一些具有特定的 python 依赖项。我不能简单地在系统级别添加所有 DAG 依赖项。当然，我可以为常见用途设置基线，但对于特定需求，依赖压缩 DAG 功能会非常有帮助。

因此，为了解决多上下文问题，我正在测试 Airflow 1.9.0（在 Ubuntu 16.04 上）的 packaged DAG 功能。

我正在按照示例使用任意 pypi 包对其进行测试。

我随机选择了一个 python 模块 (python-crontab)。（在此之前，我尝试使用更强大的模块，但重现测试需要更长的时间）
测试场景：能够导入该模块并在压缩的 DAG 中打印其版本

这是我的做法：

$ virtualenv venv --python=python3
$ source venv/bin/activate
(venv) $ mkdir contents && cd contents
$ pip install --install-option="--install-lib=$PWD" python-crontab
$ cp ../my_dag.py .
$ zip -r ../test_zip_2.zip *
$ cp ../test_zip_2.zip /path/to/dags
$ journalctl -f -u airflow-scheduler.service
(...)
WARNING - No viable dags retrieved from /path/to/dags/test_zip_2.zip

我的 DAG 的内容：

import crontab
import airflow.utils.dates as a_dates
from airflow.operators.python_operator import PythonOperator
from airflow.operators.dummy_operator import DummyOperator
from airflow.models import DAG

from pprint import pprint

args = {
    'owner': 'airflow',
    'start_date': a_dates.days_ago(1)
}

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

with DAG(dag_id='test_zip', default_args=args, schedule_interval=None) as dag:
    (
        PythonOperator(
            task_id='print_the_context',
            provide_context=True,
            python_callable=print_context,
        )
        >> DummyOperator(
            task_id='do_nothing'
        )
    )

在检查the code 后，如果找到不包含“DAG”和“airflow”字样的 .py 文件，解析 ZIP 文件的逻辑似乎会立即退出。

问题是，我上面描述的方法实际上在存档的根目录生成了其他 .py 文件。

$ ll
total 100
drwxr-xr-x 1 vagrant vagrant   442 Jun  1 14:48 ./
drwxr-xr-x 1 vagrant vagrant   306 Jun  1 15:30 ../
-rw-rw-r-- 1 vagrant vagrant  3904 Dec 30  2015 cronlog.py
-rw-rw-r-- 1 vagrant vagrant 44651 May 25 16:44 crontab.py
-rw-rw-r-- 1 vagrant vagrant  4438 Dec 28  2015 crontabs.py
drwxr-xr-x 1 vagrant vagrant   476 Jun  1 14:26 dateutil/
-rw-r--r-- 1 vagrant vagrant  6148 Jun  1 14:24 .DS_Store
drwxr-xr-x 1 vagrant vagrant   204 Jun  1 14:26 __pycache__/
drwxr-xr-x 1 vagrant vagrant   272 Jun  1 14:26 python_crontab-2.3.3-py3.5.egg-info/
drwxr-xr-x 1 vagrant vagrant   306 Jun  1 14:26 python_dateutil-2.7.3-py3.5.egg-info/
drwxr-xr-x 1 vagrant vagrant   238 Jun  1 14:26 six-1.11.0-py3.5.egg-info/
-rw-rw-r-- 1 vagrant vagrant 30888 Sep 17  2017 six.py
-rw-r--r-- 1 vagrant vagrant   832 Jun  1 14:48 my_dag.py

我测试过的许多知名包都会生成这些顶级 .py 文件。例如。安装 scrapy、numpy、pandas 等会产生同样的混乱。

那么，我有什么选择（不分叉气流^_^）？

我是否正确理解此功能？

感谢您的帮助！

【问题讨论】：

注意：PythonVirtualEnvOperator 没有解决这种情况。我的一些工作需要大量的库，并且这个操作符在每次运行时都会生成即时的一次性虚拟环境，每次执行都会增加几分钟的延迟。
对于 Terraform、Ansible、Puppet 等“常见嫌疑人”而言，安装软件包不是更多的工作，而 zip 文件仅适用于 Airflow 内容吗？
@tobi6 这就是想法：我正在使用 puppet 作为共享模块。但是对于我公司不熟悉 puppet 和其他内部操作工具的数据科学家来说，有机会在不需要 devops 帮助的情况下使用新模块测试 DAG 将是一个巨大的胜利。
在进一步思考之后，我发现可以以优雅的方式处理这些 3rd 方模块：修补 Airflow 以支持压缩 DAG 中的不同树结构。粗略的想法：pip 安装将在子文件夹中完成... zip 内容：./a_dag.py ./modules ./modules/module1/ ./modules/module2/...

标签： python airflow

【解决方案1】：

对于那些现在阅读这篇文章的人应该遵循下面的更新说明，从 v1.10.3 开始。

注意

在搜索 DAG 时，Airflow 默认只考虑包含字符串“airflow”和“DAG”的 python 文件。要考虑所有 python 文件，请禁用 DAG_DISCOVERY_SAFE_MODE 配置标志。

https://github.com/apache/airflow/blob/1.10.3/docs/concepts.rst#dags

https://github.com/apache/airflow/blob/1.10.3/UPDATING.md#new-dag_discovery_safe_mode-config-option-1

【讨论】：

【解决方案2】：

编辑：修复已合并到 1.10-stable 版本中，不应再发生。

不幸的是，从代码的当前状态来看，您想要的似乎是不可能的。

我已经在 Apache Airflow 的 GitHub 上就这个问题 here 提出了拉取请求；如果你有兴趣跟随。

【讨论】：