如何将Airflow Task Group标记为Skipped？解决组内任务跳过仍显示成功

针对你遇到的问题——Task Group因存在成功的心跳任务，即使核心数据处理任务被标记为Skipped仍显示success，导致折叠视图无法直观发现异常，可通过以下方案解决：

方案一：添加组内状态校验任务 #

在Task Group末尾新增一个校验任务，依赖组内所有任务，专门检查核心任务的状态。若核心任务被标记为Skipped，则让该校验任务失败，进而使整个Task Group状态变为failed，在折叠视图中一目了然。

代码示例 #

from airflow.models import TaskInstance
from airflow.utils.state import State
from airflow.operators.python import PythonOperator
from airflow.utils.task_group import TaskGroup

def check_core_task_status(**context):
    # 定义需要监控的核心任务ID列表
    core_task_ids = ["core_data_processing"]  # 替换为你的核心任务ID
    dag_run = context["dag_run"]
    
    for task_id in core_task_ids:
        ti = TaskInstance(task_id=task_id, dag_run=dag_run)
        ti.refresh_from_db()
        if ti.state == State.SKIPPED:
            raise ValueError(f"核心任务 {task_id} 已被标记为Skipped，触发组状态异常")

# 在Task Group中集成校验任务
with TaskGroup(group_id="data_processing_group") as processing_group:
    # 心跳检查任务
    heartbeat_task = BashOperator(
        task_id="heartbeat_check",
        bash_command="echo 'Heartbeat success'"
    )
    # 核心数据处理任务
    core_processing_task = PythonOperator(
        task_id="core_data_processing",
        python_callable=your_core_processing_function  # 替换为你的核心任务逻辑
    )
    # 状态校验任务，设置trigger_rule为all_done确保所有任务完成后执行
    status_check_task = PythonOperator(
        task_id="check_core_task_status",
        python_callable=check_core_task_status,
        provide_context=True,
        trigger_rule="all_done"
    )
    
    # 设置任务依赖：心跳和核心任务完成后执行校验
    [heartbeat_task, core_processing_task] >> status_check_task

方案说明 #

• 校验任务的trigger_rule设为all_done，确保无论核心任务是成功还是Skipped，都会执行状态检查。
• 一旦核心任务被标记为Skipped，校验任务会主动抛出异常，将自身状态设为failed，从而让整个Task Group的状态变为failed，在折叠视图中能直接识别出异常组。
• 此方案无需修改Airflow核心源码，完全通过任务编排实现，兼容性强。

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