Apache Airflow 任务级异常迁入 Task SDK从airflow.exceptions到airflow.sdk.exceptions的迁移指南【免费下载链接】airflowApache Airflow - A platform to programmatically author, schedule, and monitor workflows项目地址: https://gitcode.com/GitHub_Trending/ai/airflow导读本指南基于 Airflow 仓库中编号为54505的重大变更记录见 airflow-core/newsfragments/54505.significant.rst系统讲解 Airflow 将任务级task-facing异常类从airflow.exceptions迁移至独立 Task SDK 模块airflow.sdk.exceptions的架构调整。读者将掌握变更的动机与影响面、新旧导入路径的对应关系、airflow.exceptions代理与DeprecatedImportWarning的向后兼容机制、传感器参数校验从抛AirflowException改为抛ValueError的行为变化以及自定义 Operator、Sensor 与 Provider 的完整迁移步骤。背景为什么要把任务级异常搬进 Task SDKApache Airflow 采用Task SDK airflow-core的分层架构task-sdk目录下的独立 Python 包见 task-sdk/README.md承载任务在 Worker 上运行时所需的最小运行时依赖。变更之前任务执行期间抛出的各类信号异常跳过、失败、延迟执行、超时等定义在airflow-core的airflow.exceptions中这意味着Worker 运行时不得不依赖整个airflow-core违背了任务执行环境轻量化、与调度/元数据层解耦的目标。本次变更newsfragment54505的核心动作可以概括为三点运行时代码统一改抛 SDK 版本的异常任务实际执行路径上抛出的不再是airflow-core中的类而是airflow.sdk.exceptions中重新定义的同类Task SDK 自行定义这些异常类airflow.sdk.exceptions不再引用airflow-core从源码结构看它只依赖airflow.sdk内部的共享定义如从airflow.sdk._shared.configuration.exceptions再导出的AirflowConfigException从而让 Worker 摆脱对airflow-core的运行时依赖airflow.exceptions保留为兼容代理老的导入路径仍然可用但会发出DeprecatedImportWarning给 Dag 作者留出迁移窗口待 shim 移除后即失效。新家airflow.sdk.exceptions提供的异常清单所有任务级异常现由 task-sdk/src/airflow/sdk/exceptions.py 提供可按职责分为几类基础与通用异常异常类语义依据源码 docstringAirflowException所有 Airflow 错误的基类任何自定义异常都应继承它AirflowNotFoundException请求的对象/资源在系统中不存在AirflowOptionalProviderFeatureExceptionProvider 缺少可选功能的可选依赖时抛出AirflowDagCycleExceptionDAG 定义存在环时抛出任务状态信号类异常异常类语义AirflowSkipException任务应被标记为 skippedAirflowFailException任务应直接失败且不重试AirflowTaskTimeout任务执行超时注意继承自BaseExceptionAirflowTaskTerminated任务执行被终止继承自BaseExceptionAirflowRescheduleException任务应被安排在稍后时间重新调度携带reschedule_dateAirflowSensorTimeout传感器轮询超时源码中值得注意的细节是AirflowTaskTerminated与AirflowTaskTimeout直接继承BaseException而非AirflowException源码注释task-sdk/src/airflow/sdk/exceptions.py#L186-L195解释了原因——这类异常用于显式打断正在执行的任务普通错误处理代码不应把它们当作可正常捕获的错误类比KeyboardInterrupt因此不能落入用户except Exception的兜底分支。延迟执行Deferral相关异常异常类语义TaskDeferred算子请求将任务转入 deferred 状态、等待 Trigger 触发同样继承BaseExceptionTaskDeferralTimeout延迟等待超时TaskDeferralError任务在延迟期间因某种原因失败TaskDeferred的构造参数trigger、method_name、kwargs、timeout与serialize()方法一起定义了触发器的序列化协议用于跨进程传递延迟执行上下文task-sdk/src/airflow/sdk/exceptions.py#L197-L236。跨任务控制流信号异常类语义DagRunTriggerException算子请求触发特定 DAG 的某个 Dag RunTriggerDagRunOperator使用DownstreamTasksSkipped算子请求跳过其下游任务ShortCircuitOperator使用定义期/解析期错误异常类语义ParamValidationErrorDAG 参数校验失败多重继承AirflowException, ValueErrorDuplicateTaskIdFound同一 DAG 中定义重复task_idTaskAlreadyInTaskGroup任务已属于某 TaskGroup 又试图加入另一个TaskNotFound任务在系统中不可用XComNotFound解析 XCom 引用时对应的 XCom 不存在AirflowInactiveAssetInInletOrOutletException任务的 inlet/outlet 中包含非活跃 Asset 时执行失败向后兼容机制airflow.exceptions代理与DeprecatedImportWarning迁移后airflow.exceptions模块本身不再定义这些异常类而是通过模块级__getattr__动态代理airflow-core/src/airflow/exceptions.py#L336-L372实现兼容模块内维护了一个_DEPRECATED_EXCEPTIONS集合列出了全部被迁移的 17 个异常名AirflowDagCycleException、AirflowFailException、AirflowSensorTimeout、AirflowSkipException、TaskDeferred、ParamValidationError、XComNotFound等当代码访问airflow.exceptions.name时__getattr__命中该集合发出DeprecatedImportWarning警告内容形如airflow.exceptions.name is deprecated ... Use airflow.sdk.exceptions.name instead.stacklevel2保证警告定位到用户调用点随后通过import_string惰性加载并返回airflow.sdk.exceptions中的真实类因此老代码拿到的仍然是同一个类对象DAG 与算子可以无感继续运行。实际效果是现有 Dags/operators 继续从airflow.exceptions导入可以正常工作唯一可见的变化是日志中出现弃用警告提示迁移到 SDK 导入路径。行为变化传感器参数校验改抛ValueError本次变更还伴随一个行为层面的破坏性调整Sensors 及其他校验用户输入的辅助函数在poke_interval/timeout参数非法时由原来的抛AirflowException改为抛ValueError。这对用户的影响集中在自定义校验逻辑上如果你的代码此前通过捕获AirflowException来处理传感器参数非法的情况迁移后需要改为捕获ValueError或同时兼容两者。从类型体系看ParamValidationError本身多重继承AirflowException, ValueError说明该版本中参数校验错误与内置ValueError语义已经对齐这为上层按内置异常类型处理提供了依据。Provider 兼容层airflow.providers.common.compat.sdk对于需要同时支持 Airflow 2 与 Airflow 3 的 Provider 开发者本次变更引入了统一的兼容导入入口airflow.providers.common.compat.sdk。从 providers/common/compat/src/airflow/providers/common/compat/sdk.py 的模块文档可知该模块基于AIRFLOW_V_3_0_PLUS版本标记提供惰性导入优先尝试 Airflow 3 的新路径失败时回退到 Airflow 2 的旧路径。其配套单元测试见 providers/common/compat/tests/unit/common/compat/test_sdk.py。这意味着 Provider 只需维护一条导入路径即可在支持的多个 Airflow 版本间保持一致无需在代码里写版本分支。迁移指南1. 更新自定义 Operator / Sensor / 扩展的导入将任务级异常的导入从airflow.exceptions改为airflow.sdk.exceptions# 旧写法仍然可用但会触发 DeprecatedImportWarning from airflow.exceptions import AirflowSkipException, TaskDeferred, AirflowFailException # 新写法 from airflow.sdk.exceptions import AirflowSkipException, TaskDeferred, AirflowFailException需要迁移的常见异常包括AirflowSkipException、AirflowFailException、TaskDeferred、TaskDeferralTimeout、TaskDeferralError、AirflowSensorTimeout、AirflowRescheduleException、AirflowTaskTimeout、AirflowTaskTerminated、AirflowDagCycleException、DownstreamTasksSkipped、DagRunTriggerException、ParamValidationError、XComNotFound、DuplicateTaskIdFound、TaskAlreadyInTaskGroup、AirflowInactiveAssetInInletOrOutletException等。对于同时维护多个 Airflow 版本兼容的 Provider 代码优先走兼容层from airflow.providers.common.compat.sdk import exceptions # 按需使用其中导出的异常2. 调整传感器参数校验的异常捕获如果自定义校验逻辑此前通过捕获AirflowException识别非法的poke_interval/timeout需要改为捕获ValueError# 旧写法 try: sensor.poke_interval -1 # 非法值 except AirflowException: handle_invalid_interval() # 新写法 try: sensor.poke_interval -1 # 非法值 except ValueError: handle_invalid_interval()3. 排查存量代码升级后可以通过以下线索定位需要迁移的代码运行期日志中出现DeprecatedImportWarning: airflow.exceptions.name is deprecated ...的位置即存在旧导入代码中from airflow.exceptions import的导入语句逐一核对是否属于_DEPRECATED_EXCEPTIONS覆盖的名单。变更类型与影响范围根据 newsfragment 勾选的变更类型本次调整属于行为变更Behaviour changes与代码接口变更Code interface changes不涉及 DAG 结构、配置项、API 端点或 CLI 接口的变更。迁移规则聚焦于两条任务级异常改从airflow.sdk.exceptions导入传感器非法参数改按ValueError捕获。值得注意的是变更记录同时强调airflow.exceptions中的代理仅是为迁移提供的临时 shim未来版本会移除因此长期维护的代码应尽快切换而非长期依赖警告路径。【免费下载链接】airflowApache Airflow - A platform to programmatically author, schedule, and monitor workflows项目地址: https://gitcode.com/GitHub_Trending/ai/airflow创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考