数据可观测性：从管道监控到数据宕机的全方位保障

# dag_orders_etl.py
from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.sensors.external_task import ExternalTaskSensor

with DAG(
    dag_id="orders_etl",
    schedule="0 */2 * * *",
    start_date=datetime(2026, 9, 1),
    catchup=False,
    default_args={
        "retries": 2,
        "retry_delay": timedelta(minutes=5),
        # 失败与 SLA 未达标时推送结构化告警
        "on_failure_callback": notify_oncall,
        "sla": timedelta(hours=2),
        "email_on_retry": False,
    },
    max_active_runs=1,
) as dag:

    wait_upstream = ExternalTaskSensor(
        task_id="wait_orders_raw",
        external_dag_id="orders_raw_ingest",
        external_task_id="done",
        timeout=3600,
        soft_fail=True,
    )

    run_merge = PythonOperator(task_id="merge_orders", python_callable=merge_orders)
    build_metric = PythonOperator(task_id="build_daily_revenue", python_callable=build_daily_revenue)

    wait_upstream >> run_merge >> build_metric

# pipeline_metrics.py
from prometheus_client import Counter, Histogram, Gauge

RUN_TOTAL = Counter("pipeline_run_total", "Pipeline runs", ["dag", "task"])
RUN_FAILED = Counter("pipeline_run_failed", "Pipeline failures", ["dag", "task"])
RUN_DURATION = Histogram("pipeline_run_duration_seconds", "Pipeline duration", ["dag"])
FRESHNESS = Gauge("data_freshness_seconds", "Data freshness lag", ["table"])

def record_run(dag: str, task: str, ok: bool, duration: float):
    RUN_TOTAL.labels(dag, task).inc()
    if not ok:
        RUN_FAILED.labels(dag, task).inc()
    RUN_DURATION.labels(dag).observe(duration)

# checks/freshness.yml
checks for ods_orders:
  # 核心表新鲜度 SLO：滞后不超过 30 分钟
  - freshness using created_at < 30m:
      name: orders_freshness_slo
  # 双保险：基于最大主键时间判断
  - max(created_at) > ${{ yesterday_ts }}:
      name: orders_max_ts_recent

# freshness_controller.py
def handle_freshness_breach(table: str, lag_minutes: float):
    if lag_minutes > 30:
        # 触发上游重跑
        trigger_dag("orders_raw_ingest")
        # 通知订阅该表的团队
        notify_subscribers(table, f"freshness {lag_minutes}m exceeded SLO")
    if lag_minutes > 120:
        # 超过 2 小时：标记数据宕机，接入 on-call
        escalate_to_oncall("SEV2", table)

# volume_anomaly.py
from statistics import mean, pstdev

def check_volume(history: list, current: int, z_threshold: float = 3.5) -> dict:
    mu = mean(history)
    sigma = pstdev(history) or 1.0
    z = abs(current - mu) / sigma
    return {
        "z_score": round(z, 2),
        "baseline": round(mu, 0),
        "current": current,
        "anomaly": z > z_threshold,
    }

# 近 14 天订单量基线
history = [1_201_000, 1_210_000, 1_195_000, 1_208_000, 1_202_000,
           1_196_000, 1_210_500, 1_199_000, 1_204_000, 1_207_000,
           1_201_500, 1_198_500, 1_205_000, 1_203_000]
print(check_volume(history, current=1_205_000))   # 正常
print(check_volume(history, current=930_000))     # 异常

-- 对比两日的表结构，检测新增/删除列
SELECT column_name, data_type,
       COUNT(*) FILTER (WHERE snapshot_date = CURRENT_DATE) AS today_cnt,
       COUNT(*) FILTER (WHERE snapshot_date = CURRENT_DATE - 1) AS yesterday_cnt
FROM information_schema.columns
CROSS JOIN generate_series(CURRENT_DATE - 1, CURRENT_DATE, INTERVAL '1 day') AS snapshot_date
WHERE table_name = 'ods_orders'
GROUP BY column_name, data_type
HAVING today_cnt <> yesterday_cnt;

# assets_observability.py
from dagster import asset, AssetExecutionContext, MetadataValue

@asset(compute_kind="spark")
def daily_order_revenue(context: AssetExecutionContext):
    df = run_spark_job("daily_revenue.sql")
    # 把行数、质量分、血缘作为资产元数据暴露
    context.log_event(
        MetadataValue.int(len(df)),
        MetadataValue.text("98.2"),
        MetadataValue.url("lake.dws.daily_order_revenue"),
    )
    return df

# 资产未按时物化即触发告警
from dagster import asset_sensor, RunRequest

@asset_sensor(asset_key="daily_order_revenue")
def missing_revenue_sensor(context, event):
    return RunRequest(run_key=f"rerun-{event.asset_key}")

# sla_callback.py
def sla_miss_callback(dag, task_list, blocking_task_list, slas, *args):
    """DAG 级 SLA 违约回调：推送结构化事件到监控平台"""
    event = {
        "type": "sla_miss",
        "dag": dag.dag_id,
        "tasks": [t.task_id for t in task_list],
        "blocking": [t.task_id for t in blocking_task_list],
        "sla_date": slas[0].execution_time.isoformat(),
    }
    emit_metric("data_pipeline_sla_miss", 1, {"dag": dag.dag_id})
    notify_channel("#data-oncall", event)

# data_trace.py
from opentelemetry import trace

tracer = trace.get_tracer("data-pipeline")
with tracer.start_as_current_span("orders_freshness_check") as span:
    span.set_attribute("table", "ods_orders")
    span.set_attribute("expected_lag_min", 30)
    lag = measure_freshness("ods_orders")
    span.set_attribute("actual_lag_min", lag)
    if lag > 30:
        span.set_attribute("slo_breach", True)

{
  "ts": "2026-09-27T02:00:00Z",
  "dag": "orders_etl",
  "run_id": "scheduled__2026-09-27T02:00:00+00:00",
  "task": "merge_orders",
  "level": "WARN",
  "message": "partition dt=2026-09-26 late by 12m",
  "metrics": {"rows_processed": 1200000, "late_partitions": 2}
}

# alert_routing.yaml
routing:
  - severity: SEV1
    page: true
    escalation:
      - team: data-platform-oncall
        delay_min: 0
      - team: data-platform-leads
        delay_min: 15
  - severity: SEV2
    page: false
    channel: "#data-oncall"
    notify_owner: true
  - severity: SEV3
    channel: "#data-quality"
    business_hours_only: true
runbook_map:
  freshness: "runbook://data/freshness-breach"
  volume: "runbook://data/volume-drop"
  schema: "runbook://data/schema-change"

# oncall_handoff.py
def build_handoff(severity: str, incident: dict) -> dict:
    return {
        "title": f"[{severity}] {incident['table']} data incident",
        "summary": incident["summary"],
        "runbook": incident["runbook"],
        "affected": incident.get("downstream", []),
        "impact": incident.get("business_impact", "unknown"),
        "handoff_notes": (
            f"started_at={incident['started_at']}; "
            f"baseline={incident['baseline']}; current={incident['current']}"
        ),
    }

#!/bin/bash
# run_dbt_with_observability.sh
set -euo pipefail

# 构建并测试
dbt build --select daily_order_revenue+ --target prod

# elementary 采集测试与运行元数据
edr monitor-report

# 若关键测试失败，退出非零码触发告警
dbt test --select tag:critical --target prod

# dbt_project.yml 中的测试声明
models:
  daily_order_revenue:
    columns:
      - name: dt
        tests: [not_null, unique]
      - name: revenue
        tests:
          - not_null
          - elementary.column_anomalies:
              anomaly_direction: both
              sensitivity: 1.5

# elt_gate.py
def elt_quality_gate(model: str, check_result: dict):
    if not check_result["success"]:
        # 1. 阻断下游模型运行
        block_downstream(model)
        # 2. 标记数据宕机开始时间
        mark_data_downtime(model, started_at=datetime.utcnow())
        # 3. 通知订阅团队
        notify_subscribers(model, check_result)