Apache Airflow Listeners 完整指南基于 Pluggy 的事件监听插件开发与版本兼容实践【免费下载链接】airflowApache Airflow - A platform to programmatically author, schedule, and monitor workflows项目地址: https://gitcode.com/GitHub_Trending/ai/airflow在 Airflow 中一个工作流从调度到结束会经历 SchedulerJob 启停、DagRun 状态流转、TaskInstance 执行、Asset 数据更新、DAG 代码导入失败等大量关键事件。Listeners监听器正是 Apache Airflow 提供给开发者的一种钩子机制编写少量代码即可让 Airflow 在这些事件发生时主动通知你从而在不动业务 DAG 的前提下实现告警、审计、指标收集等横切能力。本文以仓库中的官方文档 listeners.rst 为主线结合其引用的 官方示例 listener 实现 与底层 hookspec 源码系统讲解 Airflow Listeners 支持的全部事件类型、编写方法与插件接入方式并给出跨版本兼容的实践方案。读完后你将能独立实现一个包含 DagRun / TaskInstance / Asset / 生命周期事件的监听插件。什么是 Airflow Listeners监听器允许你在 Airflow 内部发生特定事件时收到通知。它并非自研一套事件框架而是由成熟的 Python 插件库Pluggypytest 同款插件机制驱动Airflow 在内部用HookspecMarker(airflow)声明“规范”hookspec定义每个事件方法的名称与参数开发者使用hookimpl实现这些规范方法hookimplAirflow 启动时通过 ListenerManager 收集所有注册的 hookimpl 并逐一回调。从代码层面看airflow.listeners.hookimpl是从共享层重新导出的开发者只需一行导入即可使用from airflow.listeners import hookimpl使用前必须了解的警告Listeners 属于Airflow 的高级特性需要特别注意Listeners 与它们运行所在的 Airflow 组件如 Scheduler、Worker、Triggerer 等并不隔离它们直接在组件进程内执行可能拖慢、甚至在极端情况下拖垮你的 Airflow 实例。编写 listener 时应格外小心。因此编写时应当遵循两个基本原则逻辑保持轻量、耗时操作异步化任何异常都要在 listener 内部兜底捕获避免把业务组件一起带崩。Listeners 支持的事件全景根据文档Airflow 为以下五类事件提供监听能力事件类别提供的方法触发时机生命周期事件Lifecycle Eventson_starting、before_stoppingAirflowJob如SchedulerJob、Worker、Task Runner启动与停止前后DagRun 状态变更事件on_dag_run_running、on_dag_run_success、on_dag_run_failedDagRun状态发生变更时Airflow 3 起通过 API/UI 触发如 UI 上手动标记成功/失败也会收到通知TaskInstance 状态变更事件on_task_instance_running、on_task_instance_success、on_task_instance_failed、on_task_instance_skippedRuntimeTaskInstance执行期间状态变更时Airflow 3 起 API 触发的状态变更也会通知Asset 事件on_asset_created、on_asset_alias_created、on_asset_changed执行 Asset 管理操作时DAG 导入错误事件on_new_dag_import_error、on_existing_dag_import_errorDagProcessor 在 DAG 代码中发现导入错误并更新元数据库时标注为 experimental生命周期事件Lifecycle Eventson_starting before_stopping生命周期事件让你可以在一个 AirflowJob例如SchedulerJob启动和停止时做出响应。对应 hookspec 定义在共享层的 lifecycle.py 中其参数为触发该回调的组件对象hookspec def on_starting(component): Execute before Airflow component - jobs like scheduler, worker, or task runner starts. hookspec def before_stopping(component): Execute before Airflow component - jobs like scheduler, worker, or task runner stops.按规范注释的语义on_starting保证在任何其他插件方法之前被调用而before_stopping保证在之后被调用。这类事件适合做组件级的资源初始化/清理、启动心跳上报等。DagRun 状态变更事件DagRun 状态变更事件在DagRun改变状态时触发。从 Airflow 3 开始当状态变更是通过 API 触发时例如在 UI 上把某个 DagRun 标记为成功或失败listeners 同样会收到on_dag_run_success/on_dag_run_failed通知。三类方法签名一致dag_run 附带消息msg官方规范定义见 dagrun.pyhookspec def on_dag_run_running(dag_run: DagRun, msg: str): Execute when dag run state changes to RUNNING. hookspec def on_dag_run_success(dag_run: DagRun, msg: str): Execute when dag run state changes to SUCCESS. hookspec def on_dag_run_failed(dag_run: DagRun, msg: str): Execute when dag run state changes to FAIL.on_dag_run_running参考 event_listener.py 中的官方示例实现hookimpl def on_dag_run_running(dag_run: DagRun, msg: str): This method is called when dag run state changes to RUNNING. print(Dag run in running state) queued_at dag_run.queued_at version dag_run.version_number print(fDag information Queued at: {queued_at} version: {version})on_dag_run_successhookimpl def on_dag_run_success(dag_run: DagRun, msg: str): This method is called when dag run state changes to SUCCESS. print(Dag run in success state) start_date dag_run.start_date end_date dag_run.end_date print(fDag run start:{start_date} end:{end_date})on_dag_run_failedhookimpl def on_dag_run_failed(dag_run: DagRun, msg: str): This method is called when dag run state changes to FAILED. print(Dag run in failure state) dag_id dag_run.dag_id run_id dag_run.run_id run_type dag_run.run_type print(fDag information:{dag_id} Run id: {run_id} Run type: {run_type}) print(fFailed with message: {msg})可见 DagRun 事件回调中可以直接从dag_run对象读取dag_id、run_id、run_type、start_date、end_date、queued_at、version_number等字段非常适合做「整条 DAG 运行成功/失败的告警与统计」。TaskInstance 状态变更事件TaskInstance 状态变更事件在RuntimeTaskInstance改变状态时触发可用于对LocalTaskJob状态变化做出响应。文档特别说明了一个Airflow 3 的重要区别当状态变更是通过 API 触发的例如在 UI 上把某个 task instance 标记为成功或失败listener 收到的将是一个TaskInstance实例而不是RuntimeTaskInstance实例。也就是说回调参数task_instance的类型可能有两种在实现时建议用isinstance()加以区分官方示例正是这么做的。TaskInstance 相关 hookspec 定义在共享层的 taskinstance.py。on_task_instance_runninghookimpl def on_task_instance_running( previous_state: TaskInstanceState | None, task_instance: RuntimeTaskInstance | TaskInstance ): Called when task state changes to RUNNING. previous_task_state and task_instance object can be used to retrieve more information about current task_instance that is running, its dag_run, task and dag information. print(Task instance is in running state) print( Previous state of the Task instance:, previous_state) name: str task_instance.task_id context task_instance.get_template_context() task context[task] dag task.dag dag_name None if dag: dag_name dag.dag_id print(fCurrent task name:{name}) print(fDag name:{dag_name})注意task_instance.get_template_context()的使用通过渲染模板上下文可以拿到task对象进而追溯它所属的 DAG 与名称——这是 listener 中做任务级富信息的常用手段。on_task_instance_successhookimpl def on_task_instance_success( previous_state: TaskInstanceState | None, task_instance: RuntimeTaskInstance | TaskInstance ): Called when task state changes to SUCCESS. A RuntimeTaskInstance is provided in most cases, except when the tasks state change is triggered through the API. In that case, the TaskInstance available on the API server will be provided instead. print(Task instance in success state) print( Previous state of the Task instance:, previous_state) if isinstance(task_instance, TaskInstance): print(Task instances state was changed through the API.) print(fTask operator:{task_instance.operator}) return context task_instance.get_template_context() operator context[task] print(fTask operator:{operator})on_task_instance_failed失败回调相比成功多了一个error字段该字段于 Airflow 2.10.0 加入用于携带失败原因hookimpl def on_task_instance_failed( previous_state: TaskInstanceState | None, task_instance: RuntimeTaskInstance | TaskInstance, error: None | str | BaseException, ): Called when task state changes to FAILED. A RuntimeTaskInstance is provided in most cases, except when the tasks state change is triggered through the API. In that case, the TaskInstance available on the API server will be provided instead. print(Task instance in failure state) if isinstance(task_instance, TaskInstance): print(Task instances state was changed through the API.) print(fTask operator:{task_instance.operator}) if error: print(fFailure caused by {error}) return context task_instance.get_template_context() task context[task] print(Task start) print(fTask:{task}) if error: print(fFailure caused by {error})on_task_instance_skipped3.2.0 新增跳过事件的触发范围需要特别注意官方 docstring 中明确界定了边界hookimpl def on_task_instance_skipped( previous_state: TaskInstanceState | None, task_instance: RuntimeTaskInstance | TaskInstance ): Called when a task instance skips itself during execution. This hook is called only when a task has started execution and then intentionally skips itself (e.g., by raising AirflowSkipException). Note: This function will NOT cover tasks that were skipped by scheduler, before execution began, such as: - Skips due to trigger rules (e.g., upstream failures) - Skips from operators like BranchPythonOperator, ShortCircuitOperator, or similar mechanisms - Any other situation in which the scheduler decides not to schedule a task for execution For comprehensive tracking of skipped tasks, use DAG-level listeners (on_dag_run_success/on_dag_run_failed) which may have access to all task states. print(Task instance was skipped)关键结论on_task_instance_skipped只覆盖任务已开始执行后主动跳过自身例如抛出AirflowSkipException的情形由调度器在执行前就决定跳过的场景不会触发此回调包括trigger rule 导致的上游失败级联跳过、BranchPythonOperator/ShortCircuitOperator等机制产生的分支跳过、以及调度器不调度该任务的任何其他情形若需要完整追踪所有被跳过任务官方建议使用 DAG 级 listeneron_dag_run_success/on_dag_run_failed因为彼时可能已经能访问到全部任务状态。Asset 事件on_asset_created on_asset_alias_created on_asset_changedAssetAirflow 2.9 引入用于取代并增强旧的 Dataset 数据感知机制相关事件在Asset 管理操作被执行时触发可用于在数据集被创建、别名被创建、或数据内容更新时驱动后续动作。这类事件通常与 Airflow 3 的数据感知调度DAG 依赖 Asset 变化自动触发配合使用做数据链路的状态追踪。DAG 导入错误事件experimentalon_new_dag_import_error on_existing_dag_import_error当 DagProcessor 在解析 DAG 代码时发现导入错误并同步更新元数据库中的对应记录时触发。这让你可以针对「新出现的 DAG 导入错误」与「已经存在的 DAG 导入错误」分别做响应例如发送告警、更新状态面板。注意文档明确将这一类事件标记为experimental接口未来可能有调整生产环境接入前应评估风险。如何编写一个 Listener根据文档创建一个 listener 只需两步导入airflow.listeners.hookimpl为你想监听的事件实现 hookimpl 方法。Airflow 将 hookspec 规范集中定义在 spec 目录dagrun / asset / importerrors以及共享层的 listeners/spec 目录lifecycle / taskinstance。文档特别强调你的实现必须使用与 hookspec 中定义相同的命名参数。如果参数与 hookspec 不一致Pluggy 会在你尝试使用插件时抛出错误。但你不需要实现每一个方法——很多 listener 只实现其中一个或其中一部分方法。例如上面 DagRun 的三个回调都必须接收dag_run, msg两个命名参数TaskInstance 系列必须接收previous_state, task_instance失败的还需error。即便你只关心成功事件也只需实现on_dag_run_success一个方法。将 listener 接入 Airflow要让 listener 在 Airflow 安装中生效需要把它作为Airflow Plugin插件的一部分引入。最直接的做法是仿照仓库中的官方示例——该示例本身就以插件文件的形式存放于 airflow-core/src/airflow/example_dags/plugins/event_listener.py其中的每个hookimpl函数都是顶层模块级定义被 Airflow 的插件加载机制收集。把这类文件放入你的plugins_folder默认~/airflow/plugins对应的插件目录Airflow 进程启动后即会自动注册这些 hookimpl。关于插件机制的底层注册流程可以参考 listener.pyget_listener_manager()使用cache缓存单例依次调用add_hookspecs注册lifecycle、dagrun、taskinstance、asset、importerrors五组规范最后通过integrate_listener_plugins(_listener_manager)把插件中实现的方法合入 ListenerManager。之后各组件在关键节点调用对应事件即可分发到你的实现。监听范围与粒度Listener API 是面向全局设计的它监听所有 DAG、所有 Operator的事件你无法通过它监听「特定某个 DAG」产生的事件。如果你需要按 DAG 粒度做回调文档建议改用 DAG 级/任务级回调机制例如on_success_callback、on_failure_callbackDAG 级别Operator 的pre_execute、post_execute等回调。这些 API 面向特定的 DAG 作者或 Operator 创建者与全局 Listeners 互补。另外listener 中的日志与print()输出会被纳入 Airflow 的日志处理体系可作为排障依据。跨版本兼容接口演进与最佳实践Pluggy 规范带来的兼容性由于 listeners 基于 Pluggy 的 hookspec/hookimpl 机制接口演进具备一个天然特性针对旧接口编写的 listener 实现对未来版本的 Airflow 是前向兼容的接口只增不减、新增字段可被忽略反之不成立如果你的 listener 是针对新版本接口实现的它可能无法在旧版本 Airflow 上工作。如果你只针对单一 Airflow 版本部署这不是问题——按你所用版本调整实现即可。但如果你在编写可能跨多个 Airflow 版本使用的插件或扩展就必须把版本差异纳入设计。典型例子on_task_instance_failed的error字段在2.10.0才加入。若 listener 直接声明error参数它就只能运行在 2.10.0 及以上版本在旧版本上会因参数不匹配而失败。官方推荐的版本门控写法文档给出的多版本兼容实现示例利用importlib.metadata探测运行版本后按版本定义不同实现from importlib.metadata import version from packaging.version import Version from airflow.listeners import hookimpl airflow_version Version(version(apache-airflow)) if airflow_version Version(2.10.0): class ClassBasedListener: ... hookimpl def on_task_instance_failed(self, previous_state, task_instance, error: None | str | BaseException): # Handle error case here pass else: class ClassBasedListener: # type: ignore[no-redef] ... hookimpl def on_task_instance_failed(self, previous_state, task_instance): # Handle no error case here pass接口变更历史一览文档整理了自 2.8.0 引入 listeners 以来的全部接口变更Airflow 版本受影响方法变更内容2.10.0on_task_instance_failed接口中新增error字段3.0.0on_task_instance_running移除 task instance listeners 的session参数task_instance对象现在为RuntimeTaskInstance实例3.0.0on_task_instance_failed、on_task_instance_success移除 task instance listeners 的session参数task_instance在 Worker 上为RuntimeTaskInstance、在 API server 上为TaskInstance3.2.0on_task_instance_skipped向接口新增该 listener 方法基于源码的多版本适配建议结合上面示例 listener 的实现可以看出一个关键编码模式由于 Airflow 3 中 TaskInstance 类事件的对象在 Worker 与 API server 两侧类型不同在方法体内部优先用isinstance(task_instance, TaskInstance)做分支参考 event_listener.py。这样即使是同一版本的 Airflow也能优雅处理「运行期状态变更」与「API 手动标记」两种数据来源。配合previous_state参数还能进一步做「从 QUEUED → RUNNING」「从 RUNNING → FAILED」等精细化的状态迁移感知实现接近状态机级别的监控告警。小结Airflow Listeners 是一套基于 Pluggy 的全局事件钩子系统覆盖 Job 生命周期、DagRun / TaskInstance 状态流转、Asset 管理与 DAG 导入错误五类场景为告警、审计、数据血缘追踪等横向能力提供了统一入口。编写时记住三条核心原则严格对齐 hookspec 的命名参数只实现你关心的事件方法保持轻量与健壮——listener 与宿主组件共享进程异常处理与性能必须放在首位面向多版本时做好版本门控并利用isinstance(task_instance, TaskInstance)区分 API 触发的状态变更。如需继续深入建议通读 官方示例 listener 完整源码、五组 hookspec 规范定义 以及 ListenerManager 注册逻辑再结合你的实际调度场景动手实现第一个监听插件。【免费下载链接】airflowApache Airflow - A platform to programmatically author, schedule, and monitor workflows项目地址: https://gitcode.com/GitHub_Trending/ai/airflow创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
