airflow multi scheduler_airflow 多scheduler

支持版本

需要Airflow2.0以上

额外配置

应该是不需要任何的额外配置，在helm里面能看到

都是直接在deployment里面填replica数量

实现方式

多个scheduler实例主要矛盾点在如何通讯上面，airflow2用的是Mysql的行锁。同时只会有一个scheduler处理一个dag。

具体代码

jobs/scheduler_job.py

_do_scheduling() {
	self._create_dagruns_for_dags() #从dag生成dagrun
}

_create_dagruns_for_dags() {
	DagModel.dags_needing_dagruns(session)
}

1
2
3
4
5
6
7
8

models/dag.py

with_row_locks(query, of=cls, session=session, **skip_locked(session=session))
1

主要就是with_row_locks函数

这里需要注意需要在airflow.cfg里面把这个配置设置成True，并且airflow的数据库版本需要支持FOR UPDATE

启动

如果这些都符合条件的话，应该就直接可以启动多个scheduler实例。