多个气流调度器答案

【问题标题】：Multiple airflow schedulers多个气流调度器
【发布时间】：2021-04-21 07:55:54
【问题描述】：

我正在尝试安装三节点气流集群。每个节点都有气流调度器、气流工作器、气流网络服务器，还有 celery、RabbitMQ 集群和 Postgres 多主集群（用 Bucardo 实现）。软件版本：

气流 2.0.1
Postregsql 13.2
Ubuntu 20.04
pyhton 3.8.5
芹菜 4.4.7
布卡多 5.6.0
RabbitMQ 3.8.2

我在启动气流调度程序时遇到了问题。

当我启动第一个（数据库为空）时，它成功启动。但是当我在另一台机器上启动另一个调度程序时（我也尝试在同一台机器上启动），它失败了：

sqlalchemy.exc.IntegrityError: (psycopg2.errors.UniqueViolation) duplicate key value violates unique constraint "job_pkey"
DETAIL:  Key (id)=(25) already exists.

[SQL: INSERT INTO job (dag_id, state, job_type, start_date, end_date, latest_heartbeat, executor_class, hostname, unixname) VALUES (%(dag_id)s, %(state)s, %(job_type)s, %(start_date)s, %(end_date)s, %(latest_heartbeat)s, %(executor_class)s, %(hostname)s, %(unixname)s) RETURNING job.id]
[parameters: {'dag_id': None, 'state': 'running', 'job_type': 'SchedulerJob', 'start_date': datetime.datetime(2021, 4, 21, 7, 39, 20, 429478, tzinfo=Timezone('UTC')), 'end_date': None, 'latest_heartbeat': datetime.datetime(2021, 4, 21, 7, 39, 20, 429504, tzinfo=Timezone('UTC')), 'executor_class': 'CeleryExecutor', 'hostname': 'hostname', 'unixname': 'root'}]
(Background on this error at: http://sqlalche.me/e/13/gkpj)

在尝试启动几次后，调度程序最终开始工作。我假设 id 增加，然后数据成功添加到数据库中：

airflow=> select * from job order by state;
 id | dag_id |  state  |   job_type   |          start_date           |           end_date            |       latest_heartbeat        | executor_class |           hostname           | unixname 
----+--------+---------+--------------+-------------------------------+-------------------------------+-------------------------------+----------------+------------------------------+----------
 26 |        | running | SchedulerJob | 2021-04-21 07:39:22.243721+00 |                               | 2021-04-21 07:39:22.243734+00 | CeleryExecutor |                machine name  | root
 25 |        | running | SchedulerJob | 2021-04-21 07:39:14.515009+00 |                               | 2021-04-21 07:39:19.632811+00 | CeleryExecutor |                machine name  | root

日志表也有警告（如果第二个和后续调度程序成功启动）：

WARNING - Failed to log action with (psycopg2.errors.UniqueViolation) duplicate key value violates unique constraint "log_pkey"
DETAIL:  Key (id)=(40) already exists.

我明白为什么调度器无法将数据插入表中，但是它应该如何正常工作，如何启动多个调度器？官方文档告诉不需要额外的配置。希望我解释得很清楚。谢谢！

【问题讨论】：

您能告诉我们您使用的是哪个气流版本以及所有服务器中use_row_level_locking配置参数的值吗？
是的，版本气流是 2.0.1，所有服务器上的 use_row_level_locking = True
你的连接字符串是这样的postgres://user@localhost/database吗？或像这样postgres://user@host1,host2,host3/database ？
我使用的是第一个连接字符串，因为我有带有主-主复制的 Postgresql 并且位于同一台机器上。

标签： airflow airflow-scheduler

【解决方案1】：

Airflow Scheduler 和 Bucardo 之间似乎存在竞争条件。

可能最简单的解决方法是在airflow.cfg 中使用这样的连接字符串顺序查询所有服务器（所有节点上都相同）：

[core]
sql_alchemy_conn=postgresql://USER:PASS@/DB?host=node1:port1&host=node2B&host=node3

为此，您需要 sqlalchemy >= 1.3

为什么会这样

在您的调度程序和 bucardo 之间存在竞争条件，试图从不同主机的表中读取和写入数据。更改不会以应有的速度传播，并且服务器对表的写入失败。

即使您将所有节点都视为“多主节点”，让所有节点首先查看同一台服务器也可以解决此问题。如果失败，他们将使用第二个。

【讨论】：

感谢您的回复！但我不这么认为。让我解释。我正在启动第一个调度程序，然后等待它启动，然后我正在尝试启动另一个调度程序。

【解决方案2】：

我询问了 Airflow 开发人员。问题出在 Bucardo，因为它不支持

'选择...更新'：

我怀疑问题出在 Bucardo，它不支持正确锁定记录。我们寄予厚望，因为它是防止许多调度程序运行相同任务的关键保护机制。 http://airflow.apache.org/docs/apache-airflow/stable/scheduler.html#database-requirements 如果这不起作用，您将遇到重复键的问题。

谢谢！

【讨论】：