Learning: airflow
输入“/”快速插入内容
Learning: airflow
Learning: airflow
飞书用户4093月22日修改
架构
•
Scheduler: 调度器，负责决定dag run的触发和 task instance 的提交
•
Executor: 执行器，负责执行task instance, executor本身只是一串在scheduler里的控制面代码
◦
数据面可以是local process执行，也可以是celery，也可以是k8s pod
▪
worker可能是预先创建的，也可能是伴生实时创建的，视executor自己决定
◦
所有executor继承自BaseExecutor, 我们可以实现自己的executor
•
Dag file processor: dag解析器, 定时解析新的dag file，一般是scheduler进程的一部分，也可以抽出来，工作就是解析文件后更新db。
•
Dag file folder: 这个dag文件目录也非常重要，parser/scheduler/task instance 都需要能访问
•
webserver： ui
•
Metadata db: 用于存储dag file解析后的元数据和各种运行时信息
集成airflow
画板
比如模型离线训练场景：
还是很适合的，唯一的问题就是得用python写，不太好维护
•
Dag run 触发
◦
设置 schedule_interval=None，关闭按时间触发
◦
使用 TriggerDagRunOperator 来触发dag进行下一次dagrun(可以条件触发，如果训完就不触发)
◦
使用airflow trigger或rest api来进行第一次触发
•
历史训练：
◦
通过CheckpointOperatpr 获取已训练的ckpt的日期，通过 DataCheckerOperator, 进行可用样本的获取和切分，当次训练样本切分结果写入xcom，被后续operator使用
•
追新：和上面一样，没区别
•
多阶段训练：
画板
JobRunner
JobRunner 的启动对应三个命令行，都是启动对应的daemon进程
•
airflow dag-processor
•
airflow scheduler
•
airflow triggerer
•
(airflow api-server)(核心进程还有一个apiserver, 不过不属于JobRunner )
具体的操作：
•
DagProcessorJob:
◦
_run_parsing_loop：创建dag
•
SchedulerJob:
◦
_create_dagruns_for_dags：创建dag run
◦
_schedule_dag_run：调度dag run，包括创建task instance ，以及将对应的task instance设置为scheduled，可以在下一步入队被执行
◦
_critical_section_enqueue_task_instances：从 db 查出 DR.state == DagRunState.RUNNING and TI.state == TaskInstanceState.SCHEDULED 的 TaskInstance, 提交给executor入队 (queue_workload()) 【这里的critical_section是指会以select ... for update 的方式query出来，保证只有一个scheduler执行入队操作, 需要mysql 8+。多scheduler时，直接配置scheduler的deployment的replica个数就行】
•
TriggerJob:
◦
用于负责调度 Deferrable Operator，也就是触发器（比如等待文件存在、等待api响应符合预期等）
◦
Deferrable Operator 是 sensor 的升级（相当于SensorAsync），sensor是一种同步阻塞的operator/task， 会占用worker资源。但是Deferrable Operator 可以将sensor的逻辑卸载到trigger，即提交一个trigger，由专门的TriggerJob负责检测是否触发, 是异步的，无需占用worker资源。
Triggerer
简单的sensor很好理解，和普通的operator没区别，只是运行的是阻塞或轮询的代码而已。当阻塞的逻辑卸载到单独的deferred operator后（也就是trigger），看看是怎么实现的。
sensor和async sensor(trigger)的代码在接口上也没有太大区别：