如何在Airflow中自定义任务状态并配置专属显示颜色？

完全可行，按以下步骤操作就能实现：

1. 后端添加自定义状态枚举 #

找到airflow/utils/state.py里的TaskInstanceState枚举类，直接新增一行自定义状态：

class TaskInstanceState(str, Enum):
    # 保留原有所有默认状态
    SUCCESS = "success"
    FAILED = "failed"
    # ... 其他默认状态
    MYCUSTOMSTATUS = "custom"  # 你的自定义状态

由于Airflow默认把State设为TaskInstanceState的别名，之后直接用State.MYCUSTOMSTATUS就能引用这个自定义状态。

2. 给自定义状态配置前端独特颜色 #

Airflow UI的状态颜色由前端代码定义，不同版本的配置位置略有区别：

• Airflow 2.x：大多在airflow/www/static/js/dag/utils/colors.js或airflow/www/static/js/components/TaskInstance.js中，找到状态颜色映射对象，新增自定义状态的颜色配置：

const STATE_COLORS = {
    success: '#50C878',
    failed: '#FF4500',
    // ... 其他默认状态颜色
    custom: '#9370DB'  // 选一个独特颜色，比如这里用紫色
};

• 若使用2.5以上的新版本，可能需要到airflow/www/static/components/TaskInstance/TaskInstance.tsx这类TS文件中修改颜色映射，逻辑一致，新增对应键值对即可。

3. 在任务中设置自定义状态 #

使用PythonOperator时，在任务函数中拿到task_instance对象后，就能直接设置自定义状态：

from airflow.utils.state import State

def my_task_func(task_instance, **context):
    # 先执行你的业务逻辑
    # 设置自定义状态
    task_instance.set_state(state=State.MYCUSTOMSTATUS)
    return "自定义状态已设置"

my_task = PythonOperator(
    task_id='my_custom_task',
    python_callable=my_task_func,
    provide_context=True,  # 开启后才能获取task_instance对象
    dag=dag
)

注意事项 #

• 自定义状态会被持久化到Airflow的数据库中，task_instance表的state字段默认长度为50字符，足够容纳自定义状态值，无需修改数据库。
• 直接修改源码的话，Airflow升级时这些改动会被覆盖，建议用自定义插件或猴子补丁实现，避免每次升级重复修改。
• Airflow自带的重试、调度逻辑不会识别自定义状态，相关业务逻辑需要自行处理。

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