python - Airflow 工作器配置

Question

我是 Airflow 的新手。我正在尝试通过引用这篇文章使用 Celery Executor 设置 Airflow 的分布式模式 https://stlong0521.github.io/20161023%20-%20Airflow.html

在详细了解规范之前，我想确认 我在一个单独的实例上安装了 PostgreSQL .

设置的详细说明如下:

Airflow 核心/服务器计算机

python 3.5

Airflow (AIRFLOW_HOME = ~/airflow)

celery

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 (AIRFLOW_HOME = ~/airflow)

celery

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

在工作机器上运行的命令的输出:

运行 Airflow 花时:

[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//

我正在通过 中的 dag Airflow 芯机并且我已将 dag 将处理的示例数据(Excel 表)复制到同一台核心机器上。

我的 worker 日志 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 的东西使它们保持同步和最新。