数据管道测试与 CI/CD：从单元测试到数据 diff 门禁

1. 上游 Schema 变更    源表加了字段/改了类型 → 下游 NULL 激增，作业不报错
2. 时间窗口错位        时区、夏令时、事件时间 vs 处理时间 → 指标系统性偏移
3. Join 基数变化       维表重复键导致行数翻倍 → 指标虚高，聚合后看不出来

        ┌─────────────────────┐
        │  生产验证 / 影子运行  │  最慢、最真、成本最高
        ├─────────────────────┤
        │  端到端集成测试       │  容器化全链路
        ├─────────────────────┤
        │  数据断言与 diff      │  结果正确性
        ├─────────────────────┤
        │  单元测试（SQL/Python）│  最快、最便宜、数量最多
        └─────────────────────┘

# models/marts/_unit_tests.yml
unit_tests:
  - name: test_order_revenue_dedup
    model: fct_orders
    given:
      - input: ref('stg_orders')
        rows:
          - {order_id: 1, user_id: 100, amount: 20.00, status: 'paid'}
          - {order_id: 1, user_id: 100, amount: 20.00, status: 'paid'}
          - {order_id: 2, user_id: 101, amount: 5.50,  status: 'refunded'}
    expect:
      rows:
        - {order_id: 1, user_id: 100, revenue: 20.00}

# tests/test_dag_integrity.py
import pytest
from airflow.models import DagBag

@pytest.fixture(scope="session")
def dagbag():
    return DagBag(dag_folder="dags/", include_examples=False)

def test_no_import_errors(dagbag):
    assert dagbag.import_errors == {}

def test_dag_has_owner_and_tags(dagbag):
    for dag_id, dag in dagbag.dags.items():
        assert dag.default_args.get("owner"), f"{dag_id} 缺少 owner"
        assert dag.tags, f"{dag_id} 缺少 tags"

def test_tasks_have_retries(dagbag):
    for dag in dagbag.dags.values():
        for task in dag.tasks:
            assert task.retries >= 2, f"{dag.dag_id}.{task.task_id} 重试不足"

# 第一次
dbt run --select fct_orders --vars '{run_date: 2026-10-06}'
cp -r warehouse/fct_orders /tmp/run1
# 第二次（同参数）
dbt run --select fct_orders --vars '{run_date: 2026-10-06}'
diff -r /tmp/run1 warehouse/fct_orders && echo "IDEMPOTENT OK"

# docker-compose.test.yml
services:
  postgres:
    image: postgres:16-alpine
    environment:
      POSTGRES_PASSWORD: test
    ports: ["5432:5432"]
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U postgres"]
      interval: 3s
      retries: 20
  minio:
    image: minio/minio
    command: server /data --console-address ":9001"
    environment:
      MINIO_ROOT_USER: minioadmin
      MINIO_ROOT_PASSWORD: minioadmin
    ports: ["9000:9000"]
  spark:
    build: ./ci/spark
    depends_on:
      postgres: {condition: service_healthy}

from testcontainers.postgres import PostgresContainer
import psycopg2

def test_incremental_merge_dedups():
    with PostgresContainer("postgres:16-alpine") as pg:
        conn = psycopg2.connect(pg.get_connection_url())
        seed_source(conn, duplicate_ids=True)
        run_pipeline(conn)          # 执行被测管道
        assert count_rows(conn, "fct_orders") == EXPECTED_UNIQUE

models:
  - name: fct_orders
    columns:
      - name: order_id
        tests: [unique, not_null]
      - name: status
        tests:
          - accepted_values:
              values: ['paid', 'refunded', 'pending']
    tests:
      - dbt_utils.expression_is_true:
          expression: "revenue >= 0"

checks for fct_orders:
  - row_count > 0
  - row_count between 90000 and 110000
  - avg(revenue) between 18 and 26
  - duplicate_count(order_id) = 0

-- 新旧结果对账，容差 0.01
SELECT
  coalesce(a.order_id, b.order_id) AS order_id,
  a.revenue AS old_rev,
  b.revenue AS new_rev,
  abs(coalesce(a.revenue,0) - coalesce(b.revenue,0)) AS delta
FROM fct_orders_old a
FULL OUTER JOIN fct_orders_new b USING (order_id)
WHERE abs(coalesce(a.revenue,0) - coalesce(b.revenue,0)) > 0.01
   OR (a.order_id IS NULL) <> (b.order_id IS NULL);

# tests/test_contract.py
import pandera as pa
from pandera import Column, DataFrameSchema, Check

orders_schema = DataFrameSchema({
    "order_id": Column(str, Check.str_matches(r"^ORD-\d+$")),
    "amount":   Column(float, Check.ge(0)),
    "status":   Column(str, Check.isin(["paid", "refunded", "pending"])),
    "ts":       Column("datetime64[ns]"),
}, strict=True)   # strict=True：多出未声明列即失败

def test_source_contract(df):
    orders_schema.validate(df)

# .github/workflows/pipeline.yml
name: data-pipeline-ci
on:
  pull_request:
  push:
    branches: [main]

jobs:
  lint-unit:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - run: pip install -r requirements.txt
      - run: sqlfluff lint models/ --dialect snowflake
      - run: dbt parse --profiles-dir ci/
      - run: dbt test --select "tag:unit"
  integration:
    needs: lint-unit
    runs-on: ubuntu-latest
    services:
      postgres: {image: "postgres:16-alpine", env: {POSTGRES_PASSWORD: test}}
    steps:
      - uses: actions/checkout@v4
      - run: docker compose -f docker-compose.test.yml up -d --wait
      - run: pytest tests/integration -x
  build-slim:
    needs: integration
    if: github.ref == 'refs/heads/main'
    steps:
      - run: dbt build --target ci --select "state:modified+"

# 定位受影响分区
dbt ls --select fct_orders --output json | jq -r '.unique_id'
# 用上一个已知良好版本重跑指定日期区间
git checkout <good_sha>
dbt run --select fct_orders --vars '{start_date: 2026-10-01, end_date: 2026-10-05}'
git checkout main

-- 影子表与线上表按天对账
SELECT
  date_trunc('day', ts) AS d,
  count(*) AS rows_shadow,
  count(*) FILTER (WHERE s.order_id IS NULL) AS missing_in_shadow,
  count(*) FILTER (WHERE p.order_id IS NULL) AS extra_in_shadow
FROM shadow.fct_orders s
FULL OUTER JOIN prod.fct_orders p USING (order_id)
GROUP BY 1 ORDER BY 1;

Day 1: 影子表跑 1 天数据 → diff = 0
Day 2: 线上跑 10% 分区（最新一天）→ 人工核对
Day 3: 全量切换，保留旧表 7 天可回退