Apache Airflow(含MWAA)自定义Operator状态及场景实现方案问询

开箱即用状态适配性说明 #

两个场景无法完全通过原生预置状态实现所有需求，适配情况如下：

• 告警但不失败场景：无原生匹配状态。原生状态枚举中不存在「告警成功」类独立状态，仅用默认SUCCESS状态无法在UI实现橙色醒目区分，只能满足不中断下游执行的基础需求。
• 传感器数据未就绪场景：可通过原生SKIPPED状态实现核心需求。传感器探测到数据未就绪时抛出AirflowSkipException，当前任务标记为SKIPPED，下游所有依赖该任务的节点会自动跳过，整个DAG运行判定为正常结束，不会识别为流水线故障。

无源码修改权限下的自定义状态落地方案 #

Airflow核心状态枚举与调度逻辑强绑定，原生不支持用户自定义新增状态类型，托管MWAA环境也不允许修改底层state.py源码，可通过以下不修改源码的方案实现需求：

告警但不失败场景实现 #

1. 算子逻辑中捕获非致命异常后，正常返回SUCCESS状态保证下游任务正常执行，同时通过任务实例添加自定义标记：

from airflow.exceptions import AirflowException

def your_executable(context, **kwargs):
    try:
        # 你的业务逻辑
        risky_operation()
    except NonFatalWarning as e:
        # 给当前任务实例添加告警标记
        context['ti'].note = "WARN: 非致命告警 - " + str(e)
        # 可选：配置告警回调推送通知
        send_warning_notification(e)
    # 正常返回成功，不影响下游
    return

2. 开发轻量UI插件上传到MWAA的插件包中，重写任务列表页的状态渲染逻辑，检测到任务note包含WARN:前缀时，将对应任务行渲染为橙色，实现UI醒目标记的需求，该方案完全不触碰核心调度源码。

传感器数据未就绪场景优化 #

如果需要区分普通跳过和数据未就绪的跳过场景，可在抛出AirflowSkipException时添加自定义标记：

from airflow.sensors.base import BaseSensorOperator
from airflow.exceptions import AirflowSkipException

class CustomDataSensor(BaseSensorOperator):
    def poke(self, context):
        data_ready = check_source_data()
        if not data_ready:
            context['ti'].note = "DATA_NOT_READY: 源系统数据未就绪"
            raise AirflowSkipException("源数据未就绪，跳过当前运行")
        return True

同样可以通过UI插件将携带DATA_NOT_READY标记的SKIPPED状态任务渲染为你需要的专属颜色，和普通跳过任务做明确区分。 #

内容的提问来源于stack exchange，提问作者Maile Cupo