Apache Airflow 系列教程 | 第7课：执行器（Executor）体系架构

导读（Introduction）欢迎来到 Apache Airflow 源码深度解析系列的第七课。在前两课中，我们分别剖析了 Scheduler 的调度决策逻辑和 DAG 解析引擎。Scheduler 决定了"哪些任务应该运行"，解析引擎确保了"系统能看到哪些 DAG"。但还有一个关键问题：任务被 Scheduler 标记为 QUEUED 之后，实际在哪里、以何种方式执行？这就是 Executor（执行器）的职责。Executor 是 Scheduler 与 Worker 之间的桥梁——它接收调度器"入队"的任务，将其分发到实际的计算资源上执行，并将执行结果反馈给调度器。Airflow 3.x 的 Executor 体系进行了重要升级：引入了统一的Workload 抽象，将任务执行、回调执行、触发器运行统一为同一种"工作负载"概念支持Multi-Team 配置，允许不同团队使用不同的执行器执行器通过Execution API与 Task SDK 通信，实现了更清晰的职责边界从单机的LocalExecutor到分布式的CeleryExecutor和KubernetesExecutor，不同的 Executor 实现适应不同的部署规模和资源管理策略。本课将带你深入理解这个执行层的设计哲学和实现细节。学习目标（Learning Objectives）完成本课学习后，你将能够：理解 Executor 在系统架构中的角色定位——Scheduler 与实际任务执行之间的协调层掌握 BaseExecutor 的核心接口设计——heartbeat、trigger_tasks、sync、change_state 的协作流程理解 Workload 抽象体系——ExecuteTask、ExecuteCallback、RunTrigger 的统一设计分析 LocalExecutor 的实现——基于 multiprocessing 的本地并行执行了解 CeleryExecutor 和 KubernetesExecutor——分布式执行的两种主流方案掌握 ExecutorLoader 的动态加载机制——多执行器、多团队的配置与实例化具备实现自定义 Executor 的能力——理解最小实现所需的接口契约正文内容（Main Content）1. Executor 在架构中的位置1.1 整体数据流┌──────────────────────────────────────────────────────────────────┐ │ Scheduler │ │ _critical_section_enqueue_task_instances() │ │ → 将 TI 状态设为 QUEUED │ │ → 调用 executor.queue_workload(workload) │ └─────────────────────────────┬────────────────────────────────────┘ │ ▼ ┌──────────────────────────────────────────────────────────────────┐ │ Executor │ │ heartbeat() → trigger_tasks() → _process_workloads() │ │ → 将 workload 分发到实际计算资源 │ │ sync() │ │ → 收集执行结果，更新 event_buffer │ └─────────────────────────────┬────────────────────────────────────┘ │ ▼ ┌──────────────────────────────────────────────────────────────────┐ │ Worker / Execution Environment │ │ BaseExecutor.run_workload(workload) │ │ → supervise_task() / supervise_callback() │ │ → 通过 Execution API 与 Airflow Server 通信 │ └──────────────────────────────────────────────────────────────────┘1.2 Executor 解决的核心问题问题Executor 的解决方案任务在哪里执行？本地进程、远程 Worker、K8s Pod如何控制并行度？parallelism参数限制同时运行的 workload 数如何获取执行结果？sync()方法轮询结果，写入event_buffer如何处理失败？状态变更通知 Scheduler，由 Scheduler 决定重试如何支持不同团队？Multi-Team 配置，每个团队可有独立 Executor2. BaseExecutor：执行器的核心抽象2.1 类定义与关键属性# 源码位置：airflow-core/src/airflow/executors/base_executor.pyclassBaseExecutor(LoggingMixin):""" Base class to inherit for concrete executors such as Celery, Kubernetes, Local, etc. """# 类级别的能力声明supports_ad_hoc_ti_run:bool=False# 是否支持即席运行supports_callbacks:bool=False# 是否支持回调执行supports_multi_team:bool=False# 是否支持多团队is_local:bool=False# 是否为本地执行器is_production:bool=True# 是否适用于生产环境serve_logs:bool=False# 是否提供日志服务pre_assigns_external_executor_id:ClassVar[bool]=False# 是否预分配外部IDdef__init__(self,parallelism:int=PARALLELISM,team_name:str|None=None):self.parallelism:int=parallelism self.team_name:str|None=team_name# 核心数据结构self.queued_tasks:dict[TaskInstanceKey,workloads.ExecuteTask]={}self.queued_callbacks:dict[str,workloads.ExecuteCallback]={}self.running:set[WorkloadKey]=set()self.event_buffer:dict[WorkloadKey,EventBufferValueType]={}self.conf=ExecutorConf(team_name)关键属性解读：属性类型说明queued_tasksdictScheduler 已入队但尚未发送到 Worker 的任务queued_callbacksdict已入队的回调工作负载runningset已发送到 Worker 正在执行的 workload keyevent_bufferdict执行完成的结果缓冲区，等待 Scheduler 读取parallelismint最大并行执行数2.2 核心方法生命周期Executor 的核心方法形成了一个清晰的生命周期：# 1. 启动defstart(self):"""Executors may need to get things started."""# 2. Scheduler 入队工作负载defqueue_workload(self,workload:ExecutorWorkload,session:Session)-None:ifisinstance(workload,workloads.ExecuteTask):self.queued_tasks[ti.key]=workloadelifisinstance(workload,workloads.ExecuteCallback):self.queued_callbacks[workload.callback.id]=workload# 3. 心跳驱动执行（被 Scheduler 循环调用）defheartbeat(self)-None:open_slots=self.parallelism-len(self.running)self._emit_metrics(open_slots,...)self.trigger_tasks(open_slots)self.sync()# 4. 触发任务执行deftrigger_tasks(self,open_slots:int)-None:workloads_to_schedule=self._get_workloads_to_schedule(open_slots)ifworkload_list:self._process_workloads(workload_list)# 子类必须实现# 5. 同步执行状态（子类覆盖）defsync(self)-None:"""Executors should override this to gather statuses."""# 6. Scheduler 读取执行结果defget_event_buffer(self,dag_ids=None)-dict[WorkloadKey,EventBufferValueType]:cleared_events=self.event_buffer self.event_buffer={}returncleared_events# 7. 关闭defend(self)-None:"""Wait synchronously for previously submitted jobs to complete."""defterminate(self):"""Called when the daemon receives a SIGTERM."""2.3 调度优先级策略def_get_workloads_to_schedule(self,open_slots:int)-list[tuple[WorkloadKey,ExecutorWorkload]]:""" Priority Policy: Callbacks are scheduled before tasks. Callbacks complete existing work; tasks start new work. """workloads_to_schedule=[]# 优先级1：回调先于任务（完成现有工作优先于开始新工作）ifself.queued_callbacks:forkey,workloadinself.queued_callbacks.items():iflen(workloads_to_schedule)=open_slots:breakworkloads_to_schedule.append((key,workload))# 优先级2：任务按 priority_weight 排序ifopen_slotslen(workloads_to_schedule)andself.queued_tasks:fortask_key,task_workloadinself.order_queued_tasks_by_priority():iflen(workloads_to_schedule)=open_slots:breakworkloads_to_schedule.append((task_key,task_workload))returnworkloads_to_scheduledeforder_queued_tasks_by_priority(self):"""Sort tasks by priority_weight (lower number = higher priority)."""returnsorted(self.queued_tasks.items(),key=lambdax:x[1].ti.priority_weight,reverse=False,)2.4 状态变更与事件缓冲defchange_state(self,key:WorkloadKey,state:WorkloadState,info=None,remove_running=True):"""Change state of the task and update event buffer."""ifremove_running:try:self.running.remove(key)exceptKeyError:passself.event_buffer[key]=state,infodeffail(self,key:WorkloadKey,info=