我正在为数据科学家团队建立一个多用户气流集群,其中包括DAG(ETL,NLP,ML,NN ……)的各种用法,其中一些具有特定的python依赖性.
我不能简单地在系统级添加所有DAG依赖项.当然,我可以设置常用的基线,但是对于特定的需求,依赖于压缩的DAG功能会非常有帮助.

因此,为了解决这个多上下文问题,我正在测试Airflow 1.9.0的packaged DAG功能(在Ubuntu 16.04上).

我正在按照这个例子用任意的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等产生同样的混乱.

那么,可能是我的选择(没有分流气流^ _ ^)？

我是否正确理解此功能？

谢谢你的帮助！

解决方法:

编辑：修复已合并到版本1.10稳定,应该不再发生.

不幸的是,从代码的当前状态看,你想要的东西是不可能的.

我已经在Apache Airflow的GitHub中对此问题here提出了拉取请求;如果你有兴趣跟随.