如何在虚拟环境中使用 apache 气流？

Question

我对使用 apache 气流很陌生。我使用 pycharm 作为我的 IDE。我创建了一个项目（anaconda 环境），创建了一个包含 DAG 定义和 Bash 运算符的 python 脚本。当我打开气流网络服务器时，我的 DAGS 未显示。仅显示默认示例 DAG。我的AIRFLOW_HOME 变量包含~/airflow。所以我在那里存储了我的python脚本，现在它显示了。

如何在项目环境中使用它？

我是否在每个项目开始时更改环境变量？

有没有办法为每个项目添加特定的气流主目录？

我不想将我的 DAG 存储在默认的气流目录中，因为我想将它添加到我的 git 存储库中。请帮帮我。

Answer 1

编辑airflow.cfg文件并设置：

load_examples = False
dags_folder = /path/to/your/dag/files

如果您的气流目录未设置为默认值，则应设置此环境变量。如果每次都更改它很烦人，只需在您的 pycharm 项目配置或本地操作系统（~/.bashrc）中设置它。

我的建议是编写一个小脚本以本地模式执行气流。 您应该单独执行气流网络服务器和调度程序和工作人员。

对我来说，在小型开发机器上运行气流服务要方便得多，考虑这样做。

Answer 2

您可以使用以下格式使用环境变量设置/覆盖${AIRFLOW_HOME}/airflow.cfg 中指定的气流选项：$AIRFLOW__{SECTION}__{KEY}（注意双下划线）。这是气流文档的link。所以你可以简单地做

export AIRFLOW__CORE__DAGS_FOLDER=/path/to/dags/folder

但是，为不同的项目执行此操作既乏味又容易出错。作为替代方案，您可以考虑使用 pipenv 代替 Anaconda 来管理虚拟环境。这是关于pipenv 及其解决的问题的nice guide。 default features 的 pipenv 之一是当您生成一个激活了 virtualenv 的 shell 时，它会自动加载在 .env 文件中定义的变量。因此，您使用pipenv 的工作流程如下所示：

cd /path/to/my_project

# Creates venv with python 3.7 
pipenv install --python=3.7 Flask==1.0.3 apache-airflow==1.10.3

# Set home for airflow in a root of your project (specified in .env file)
echo "AIRFLOW_HOME=${PWD}/airflow" >> .env

# Enters created venv and loads content of .env file 
pipenv shell

# Initialize airflow
airflow initdb
mkdir -p ${AIRFLOW_HOME}/dags/

注意：Flask==1.03的用法我会在最后解释，但这是因为pipenv会检查子依赖是否兼容，以确保可重复性。

所以在这些步骤之后，您将获得以下项目结构

my_project
├── airflow
│   ├── airflow.cfg
│   ├── airflow.db
│   ├── dags
│   ├── logs
│   │   └── scheduler
│   │       ├── 2019-07-07
│   │       └── latest -> /path/to/my_project/airflow/logs/scheduler/2019-07-07
│   └── unittests.cfg
├── .env
├── Pipfile
└── Pipfile.lock

现在，当您第一次初始化气流时，它将创建 ${AIRFLOW_HOME}/airflow.cfg 文件并将使用/扩展 ${AIRFLOW_HOME}/dags 作为 dags_folder 的值。如果您仍然需要 dags_folder 的其他位置，您可以再次使用 .env 文件

echo "AIRFLOW__CORE__DAGS_FOLDER=/different/path/to/dags/folder" >> .env

因此，您的.env 文件将如下所示：

AIRFLOW_HOME=/path/to/my_project/airflow
AIRFLOW__CORE__DAGS_FOLDER=/different/path/to/dags/folder

我们取得了哪些成就以及为什么可以正常工作

1. 由于您在虚拟环境中安装了airflow，您需要激活它才能使用airflow
2. 由于您使用pipenv 完成此操作，因此您需要使用pipenv shell 来激活venv
3. 由于您使用pipenv shell，您将始终将.env 中定义的变量导出到您的venv。除此之外，pipenv 仍将是一个子 shell，因此，当您退出它时，所有其他环境变量也会被清除。
4. 使用气流的不同项目的日志文件等会有不同的位置。

关于 pipenv 的补充说明

1. 要使用通过 pipenv 创建的 venv 作为 IDE 的项目解释器，请使用 pipenv --py 提供的路径。
2. 默认情况下，pipenv 会像 conda 一样在同一个全局位置创建所有 venv，但您可以通过将 export PIPENV_VENV_IN_PROJECT=1 添加到您的 .bashrc（或其他 @ 987654354@)。然后，当您进入项目解释器的设置时，PyCharm 将能够自动获取它。

注意Flask==1.0.3的用法

来自 PyPi 的 Airflow 1.10.3 依赖于 flask>=1.0, <2.0 和 jinja2>=2.7.3, <=2.10.0。今天，当我测试代码 sn-ps 时，最新可用的 flask 是 1.1.0，它取决于 jinja2>=2.10.1。这意味着虽然 pipenv 可以安装所有需要的软件，但它无法锁定依赖项。因此，为了干净地使用我的代码示例，我必须指定 flask 的版本，它需要与气流要求兼容的 jinja2 版本。但没有什么可担心的。 GitHub 上最新版本的airflow 已经修复了这个问题。