企业官网建设流程全解析

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 插件库Pluggy（pytest 同款插件机制）驱动：

• Airflow 在内部用HookspecMarker("airflow")声明“规范”（hookspec），定义每个事件方法的名称与参数；
• 开发者使用@hookimpl实现这些规范方法（hookimpl）；
• Airflow 启动时通过 ListenerManager 收集所有注册的 hookimpl 并逐一回调。

从代码层面看，airflow.listeners.hookimpl是从共享层重新导出的，开发者只需一行导入即可使用：

from airflow.listeners import hookimpl

使用前必须了解的警告

Listeners 属于Airflow 的高级特性，需要特别注意：

Listeners 与它们运行所在的 Airflow 组件（如 Scheduler、Worker、Triggerer 等）并不隔离，它们直接在组件进程内执行，可能拖慢、甚至在极端情况下拖垮你的 Airflow 实例。编写 listener 时应格外小心。

因此编写时应当遵循两个基本原则：逻辑保持轻量、耗时操作异步化；任何异常都要在 listener 内部兜底捕获，避免把业务组件一起带崩。

Listeners 支持的事件全景

根据文档，Airflow 为以下五类事件提供监听能力：

生命周期事件（Lifecycle Events）

on_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.py：

@hookspec 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(f"Dag information Queued at: {queued_at} version: {version}")

on_dag_run_success

@hookimpl 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(f"Dag run start:{start_date} end:{end_date}")

on_dag_run_failed

@hookimpl 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(f"Dag information:{dag_id} Run id: {run_id} Run type: {run_type}") print(f"Failed 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_running

@hookimpl 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(f"Current task name:{name}") print(f"Dag name:{dag_name}")

注意task_instance.get_template_context()的使用：通过渲染模板上下文可以拿到task对象，进而追溯它所属的 DAG 与名称——这是 listener 中做任务级富信息的常用手段。

on_task_instance_success

@hookimpl 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 task's 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 instance's state was changed through the API.") print(f"Task operator:{task_instance.operator}") return context = task_instance.get_template_context() operator = context["task"] print(f"Task 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 task's 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 instance's state was changed through the API.") print(f"Task operator:{task_instance.operator}") if error: print(f"Failure caused by {error}") return context = task_instance.get_template_context() task = context["task"] print("Task start") print(f"Task:{task}") if error: print(f"Failure caused by {error}")

on_task_instance_skipped（3.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 级 listener（on_dag_run_success/on_dag_run_failed），因为彼时可能已经能访问到全部任务状态。

Asset 事件

on_asset_created on_asset_alias_created on_asset_changed

Asset（Airflow 2.9+ 引入，用于取代并增强旧的 Dataset 数据感知机制）相关事件在Asset 管理操作被执行时触发，可用于在数据集被创建、别名被创建、或数据内容更新时驱动后续动作。这类事件通常与 Airflow 3 的数据感知调度（DAG 依赖 Asset 变化自动触发）配合使用，做数据链路的状态追踪。

DAG 导入错误事件（experimental）

on_new_dag_import_error on_existing_dag_import_error

当 DagProcessor 在解析 DAG 代码时发现导入错误，并同步更新元数据库中的对应记录时触发。这让你可以针对「新出现的 DAG 导入错误」与「已经存在的 DAG 导入错误」分别做响应（例如发送告警、更新状态面板）。注意文档明确将这一类事件标记为experimental，接口未来可能有调整，生产环境接入前应评估风险。

如何编写一个 Listener

根据文档，创建一个 listener 只需两步：

1. 导入airflow.listeners.hookimpl；
2. 为你想监听的事件实现 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.py：get_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_callback（DAG 级别）；
• 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 以来的全部接口变更：

基于源码的多版本适配建议

结合上面示例 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 导入错误五类场景，为告警、审计、数据血缘追踪等横向能力提供了统一入口。编写时记住三条核心原则：

1. 严格对齐 hookspec 的命名参数，只实现你关心的事件方法；
2. 保持轻量与健壮——listener 与宿主组件共享进程，异常处理与性能必须放在首位；
3. 面向多版本时做好版本门控，并利用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