Airflow DAG 从哪里获得它的依赖关系？

Question

我正在努力使用 Airflow 执行一些非常简单的任务。

对于上下文，我使用 docker-compose 来运行带有 Airflow 和 Postgres 的 docker 容器。 (https://github.com/puckel/docker-airflow)

我正在尝试测试我们的一个内部库与 Airflow 的集成。我用来快速测试的不是很干净的方法是将 docker exec 放入气流容器并 pip 安装适当的库（通过主机共享到具有只读模式的 Docker 卷的容器）。

所有东西都用 pip 正确安装，我可以在运行虚拟 Python 脚本时使用我的库。

但是，当我在 DAG python 文件中集成相同的逻辑时，我收到错误“损坏的 dag，没有名为 inhouse_lib 的模块。

起初我认为 Airflow 是在相对于 Python 版本的特定 pip 目录中选择依赖项，并且我将库安装在另一个 pip 目录中。

但对于所有 Python 二进制文件，它们都使用 Python 3.7。

对于我在创建 pip 列表时拥有的所有 pip 二进制文件（pip、pip3、pip3.7），我可以找到我的内部库。

我不明白我应该如何部署我的库以便 Airflow 可以接收它们。任何见解都将不胜感激。

感谢您的帮助。

编辑 为了澄清我试图做的事情，下面有一些细节。在我的 DAG 中，我想使用一个自定义 Python 库（我们称其为尚未实现的 myLib 功能。一旦实现，我想将这个最新版本的 myLib 部署到气流容器中。

我用一个卷更新了 docker-compose.yml，该卷将我的主机目录与容器气流主页上的 myLib 映射。

# Go in the container
docker exec -it <airflow docker container ID> bash

# Install myLib to Python environment
pip install myLib

# Check the installation
pip list | grep myLib # output myLib

# Check the import in Python REPL
python
import myLib # No Python error

相同的导入在我的 Airflow DAG 中不起作用。检查容器日志时，出现以下错误：

[2019-08-30 15:14:30,499] {{__init__.py:51}} INFO - Using executor LocalExecutor
[2019-08-30 15:14:30,894] {{dagbag.py:90}} INFO - Filling up the DagBag from /usr/local/airflow/dags
[2019-08-30 15:14:30,897] {{dagbag.py:205}} ERROR - Failed to import: /usr/local/airflow/dags/mydag.py
Traceback (most recent call last):
  File "/usr/local/lib/python3.7/site-packages/airflow/models/dagbag.py", line 202, in process_file
    m = imp.load_source(mod_name, filepath)
  File "/usr/local/lib/python3.7/imp.py", line 171, in load_source
    module = _load(spec)
  File "<frozen importlib._bootstrap>", line 696, in _load
  File "<frozen importlib._bootstrap>", line 677, in _load_unlocked
  File "<frozen importlib._bootstrap_external>", line 728, in exec_module
  File "<frozen importlib._bootstrap>", line 219, in _call_with_frames_removed
  File "/usr/local/airflow/dags/mydag.py", line 7, in <module>
    import myLib
ModuleNotFoundError: No module named 'myLib'
[2019-08-30 15:14:31 +0000] [167] [INFO] Handling signal: ttou
[2019-08-30 15:14:31 +0000] [11446] [INFO] Worker exiting (pid: 11446)

Answer 1

对于每个 dag，您需要在运行前对其进行测试。

您可以使用以下 cli 命令来检查环境和代码逻辑：

airflow list dags
airflow test [dag_name] [task_name] [date]

根据您的问题，您应该面临环境依赖性问题。您可以通过 docker 容器中的气流列表 dags 来检查它。为了解决您的任务，我们有两种方法： 1.在airflow.cfg文件中设置dags文件夹，把你的module文件放到dags文件夹中。

1. 检查气流环境中的python路径，确保您的模块可以访问。

=========================== 更新1： 为了检查您的模块是否正确安装，您可以使用以下命令：

1. 码头图片 | grep [你的气流图像名称]
2. 找到容器ID
3. docker run [容器 id] python
4. 在 docker python 环境中检查您的模型是否正确安装。 like : import os 如果您收到任何错误消息，您 需要通过 pip 双重检查您的模块安装。

更新 2： 为了检查您的依赖性，您可以： 编写一个简单的 dags，然后使用气流测试 [dag_name] [dag_task_name] [date] 看看是否工作。

据我了解，您可以尝试从零开始构建气流图像，它可能会很好地工作。 docker 容器可能会假设一些环境。和用户实例。

如果您更喜欢继续使用网络 posed 容器，您可以尝试在登录时切换到相同的使用 id 并安装您的 python 库。喜欢跟随模式

docker exec -u [user_name you can find you the dockerfile] [container_id] you command.

不要忘记将每次更改提交到新图像 ID 并从新图像 ID 加载容器，否则每次运行时可能会丢失更改

Answer 2

使用以下内容构建 puckel dockerfile：

docker build --rm --build-arg PYTHON_DEPS="flask_oauthlib>=0.9" -t puckel/docker-airflow .

将要通过 pip 安装的 Python 依赖项列表添加到 PYTHON_DEPS 变量中，以逗号分隔的列表形式。

这将构建安装了依赖项的映像，然后您可以在 dags 中使用它，只需 import yourpackage。