airflow调度方案

妖狐艹你老母 2022-09-07 12:25 169阅读 0赞

## 主要服务 ##

1.  `dag`：包括 task、operator、sensor 模块 配置 dag、task、operator
2.  `schedule`：包括 control 模块 启动 暂停 恢复等 控制dag任务启动
3.  `worker`：执行 task 触发sensor状态更新

## operator ##

**operator（操作器）：定义任务该以哪种方式执行**  
主要有以下四种

1.  `function_operator`：js函数
2.  `shell_operator` ：执行shell命令
3.  `http_operator`：发送http请求
4.  `seneca_operator`：发送seneca请求

所有的operator有 `default_args`（默认参数） 和 `dynamic_args`（动态参数） 和 `schedule_args` 三种参数  
**优先级**： schedule\_args > dynamic\_args > default\_args

`schedule_args` 从schedule start启动入口传入到第一批task 然后依次往后续的task传递

`dynamic_args`支持2种类型：

1.  从指定的url获取 get
2.  从指定的seneca 接口获取

default\_args 创建task的operator时输入

## sensor ##

创建task时系统自动创建`sensor`，sensor是task任务可执行的感知器 ，task可执行满足2个条件

1.  所有前向任务执行完毕
2.  任务自身的条件满足

任务自身的条件在创建task时指定，sensor支持多种条件：

*  到达指定时间范围
 *  指定url get返回结果满足要求
 *  调用seneca接口返回结果满足要求

## schedule ##

schedule：调度器  
支持接口：

*  start 多种条件启动dag任务
 *  stop 停止
 *  pause 暂停调度
 *  resume 恢复调度

start 支持多个参数

*  调度策略参数

调度策略：

1.  指定时间执行 周期间隔执行固定次数
2.  立即执行
3.  监听http事件触发执行
4.  监听seneca事件触发执行

执行参数策略：

1.  通过start接口传入固定使用
2.  指定url每次执行返回
3.  指定seneca接口每次执行返回

dag调度后 从数据库 load到redis，通过多个worker执行

worker要设置task状态，通知相应的sensor，判断sensor条件是否满足

task的状态：

1.  暂停 任务暂停，不调度
2.  调度中 任务可执行，等待worker执行
3.  执行中 分配到worker，worker正在执行任务
4.  等待 sensor条件未满足，任务等待
5.  结束 任务执行完毕
6.  异常 任务执行异常

schedule start 满足条件调度后设置DAG起点任务为等待  
shecudle中调度器判断所有等待任务，如果sensor条件满足，分配worker设置状态为调度中  
worker查找分配给自身的task，执行完毕设置状态为结束和异常，并触发task对应的下游task 为等待

task\_context 固定包含 pre\_result (前向任务执行结果集合) sche\_parameter： 执行参数  
sensor\_result： sensor触发结果

sensor的动态参数 和opertor的动态参数 $.sche\_parameter.status 这种方式

schedule调度进程 判断sensor条件满足后，触发动态参数解析，完成后设置实际参数到redis对应的task缓存中 然后worker调度执行

还没有评论，来说两句吧...