编辑“︁Airflow任务状态监控”︁（章节）

== 高级监控技巧 ==

=== 自定义指标 ===
集成Prometheus暴露监控指标：

<syntaxhighlight lang="python">
from airflow import DAG
from airflow.models import TaskInstance
from prometheus_client import Gauge

success_gauge = Gauge('airflow_task_success', 'Successful task runs')

def task_success_callback(context):
    success_gauge.inc()

with DAG(
    'monitored_dag',
    user_defined_macros={
        'on_success_callback': task_success_callback
    }
) as dag:
    # 任务定义...
</syntaxhighlight>

=== 状态历史分析 ===
查询任务状态历史记录：

<syntaxhighlight lang="sql">
-- 在Airflow元数据库中执行
SELECT task_id, state, execution_date 
FROM task_instance 
WHERE dag_id = 'my_dag'
ORDER BY execution_date DESC
LIMIT 100;
</syntaxhighlight>