Airflow 工作器配置

Question

我是 Airflow 的新手。我正在尝试通过参考这篇文章 https://stlong0521.github.io/20161023%20-%20Airflow.html

使用 Celery Executor 设置气流的分布式模式

在详细了解规范之前，我想确认 我已经在单独的实例上安装了 PostgreSQL。

设置的详细说明如下：

气流core/server电脑

Python 3.5
- 气流（AIRFLOW_HOME = ~/气流）
- 芹菜
- psycogp2
RabbitMQ

airflow.cfg中的配置：

sql_alchemy_conn = postgresql+psycopg2://username:password@192.168.2.12:5432/airflow
executor = CeleryExecutor
broker_url = amqp://username:password@192.168.1.12:5672//
celery_result_backend = db+postgresql://username:password@192.168.2.12:5432/airflow

执行的测试：

RabbitMQ is running
Can connect to PostgreSQL and have confirmed that Airflow has created tables
Can start and view the webserver (including custom dags)

Airflow worker 计算机

安装了以下软件：

Python 3.5 与
- 气流（AIRFLOW_HOME = ~/气流）
- 芹菜
psycogp2

在airflow.cfg中所做的配置与服务器中的完全相同：

sql_alchemy_conn = postgresql+psycopg2://username:password@192.168.2.12:5432/airflow
executor = CeleryExecutor
broker_url = amqp://username:password@192.168.1.12:5672//
celery_result_backend = db+postgresql://username:password@192.168.2.12:5432/airflow

worker 机器上命令运行的输出：

当运行宁气流花：

[2018-02-19 14:58:14,276] {__init__.py:57} INFO - Using executor CeleryExecutor
[2018-02-19 14:58:14,360] {driver.py:120} INFO - Generating grammar tables from /usr/lib/python3.5/lib2to3/Grammar.txt
[2018-02-19 14:58:14,384] {driver.py:120} INFO - Generating grammar tables from /usr/lib/python3.5/lib2to3/PatternGrammar.txt
[I 180219 14:58:15 command:139] Visit me at http://0.0.0.0:5555
[I 180219 14:58:15 command:144] Broker: amqp://username:password@192.168.1.12:5672//
[I 180219 14:58:15 command:147] Registered tasks: 
    ['celery.accumulate',
     'celery.backend_cleanup',
     'celery.chain',
     'celery.chord',
     'celery.chord_unlock',
     'celery.chunks',
     'celery.group',
     'celery.map',
     'celery.starmap']
[I 180219 14:58:15 mixins:224] Connected to amqp://username:password@192.168.1.12:5672//

我在 Airflow Core 机器 中传递 dag，并且我已经复制了 dag 将处理的示例数据（Excel 张）到同一个核心机.

我的工人日志 raise CalledProcessError(retcode, cmd) subprocess.CalledProcessError: Command 'airflow run dag_name_x task_name_xx 2018-02-19T10:15:41.657243 --local -sd /home/Distributedici/airflow/dags/sample_data_xx.py' returned non-zero exit status 1

现在我的查询是

1) 我是否也应该将 dag 文件夹复制到工作计算机

2) 现在，我还没有复制工作计算机上的 dag 文件夹，我无法看到工作进程接收任务。

请指出我哪里出错了，以及如何让工作进程接管进程。

Answer 1

您的配置文件看起来没问题。正如您所怀疑的那样，所有工作人员确实需要 DAG 文件夹的副本。您可以使用 git 之类的东西使它们保持同步和最新。

Answer 2

Airflow 的一些最大痛点在于部署和保持 DAG 文件和插件在您的 Airflow 调度程序、Airflow 网络服务器和 Celery 工作节点之间同步。

我们创建了一个名为 Astronomer Open 的开源项目，该项目将 Dockerized Airflow、Celery 和 PostgreSQL 与其他一些好东西一起自动化。该项目的动机是看到这么多人遇到了相同的痛点创建一个非常相似的设置。

例如，这里是 Airflow Dockerfile：https://github.com/astronomer/astronomer/blob/master/docker/airflow/1.10.2/Dockerfile

以及文档：https://open.astronomer.io/

完全披露：这是我在工作中参与的一个项目——我们提供付费 enterprise edition as well that runs on Kubernetes (docs)。也就是说，开放版是完全免费的。

Answer 3

这有点晚了，但它可能仍然对某些人有帮助，因为从现有的答案看来，除了 "manual" 部署（通过 git/scp 等）之外，没有其他方法可以共享 DAG。 ), 而有一种方法。

Airflow 支持 pickling（来自 CLI 的 -p 参数或 docker-compose 文件中的 command: scheduler -p），它允许在 server/master，并将它们序列化并发送给工作人员（这样您就不必在多个地方部署 DAG，并且可以避免 DAG 不同步的问题）。

Pickling 与 CeleryExecutor 兼容。

Pickling 有一些限制，可能会反序列化，特别是类的实际代码和函数未序列化（只有完全限定名称），因此如果您尝试反序列化，将会出现错误DAG 指的是您在目标环境中没有的代码。有关泡菜的更多信息，您可以在这里查看：https://docs.python.org/3.3/library/pickle.html

Answer 4

是的，dags 必须出现在所有气流节点上 - worker、webserver、scheduler。

您可以有一个运行 git 的 cron 拉入所有节点上的 dags 文件夹以保持同步。

Airflow 会将所有 dag 移动到数据库而不是文件系统 - 此功能可能会出现在 2.0 中。

Airflow 工作器配置

Airflow Worker Configuration

python

python-3.x

airflow

airflow-scheduler