数据质量监控实战：从质量维度到 Great Expectations 与 Soda 落地

# quality_tier.yaml
tiers:
  - name: critical        # 直接支撑财报 / 监管 / 在线推荐
    completeness: 100.0
    freshness_minutes: 5
    action: BLOCK_AND_PAGE
  - name: important       # 支撑日常运营报表
    completeness: 99.99
    freshness_minutes: 30
    action: BLOCK_AND_ALERT
  - name: nice_to_have    # 探索性分析
    completeness: 95.0
    freshness_minutes: 720
    action: ALERT_ONLY

# init_gx.py
import great_expectations as gx

context = gx.get_context(mode="file")  # 或 gx.get_context(project_config=...)

datasource = context.sources.add_or_update_sql_datasource(
    name="orders_db",
    connection_string="postgresql+psycopg2://user:pass@dw-host:5432/warehouse",
)

# 为 orders 表创建 asset
table_asset = datasource.add_table_asset(
    name="orders_asset",
    table_name="ods_orders",
    schema_name="ods",
)

# suite_orders.py
from great_expectations import gx

context = gx.get_context(mode="file")
suite = context.add_expectation_suite("ods_orders.suite")

# 主键唯一性：order_id 必须唯一
suite.expectation_context.add_expectation(
    expect_column_values_to_be_unique(column="order_id")
)

# 完整性：核心字段不允许为空
suite.expectation_context.add_expectation(
    expect_column_values_to_not_be_null(column="order_id")
)
suite.expectation_context.add_expectation(
    expect_column_values_to_not_be_null(column="user_id", mostly=0.9999)
)

# 准确性/业务规则：金额必须为正
suite.expectation_context.add_expectation(
    expect_column_values_to_be_between(column="amount", min_value=0, max_value=1_000_000)
)

# 有效性：状态码在合法枚举内
suite.expectation_context.add_expectation(
    expect_column_values_to_be_in_set(
        column="status",
        value_set=["CREATED", "PAID", "SHIPPED", "CANCELLED", "REFUNDED"],
    )
)

# 时效性兜底：当日数据量下限
suite.expectation_context.add_expectation(
    expect_table_row_count_to_be_between(min_value=100_000, max_value=50_000_000)
)

# run_checkpoint.py
import great_expectations as gx

context = gx.get_context(mode="file")

checkpoint = context.add_or_update_checkpoint(
    name="ods_orders.daily_check",
    validations=[
        {
            "batch_request": {
                "datasource_name": "orders_db",
                "data_asset_name": "orders_asset",
            },
            "expectation_suite_name": "ods_orders.suite",
        }
    ],
    result_format="COMPLETE",
)

result = context.run_checkpoint(checkpoint_name="ods_orders.daily_check")

# 结果判定：全部通过则 success=True
print("SUCCESS" if result.success else "FAILED", {
    e["expectation_config"]["expectation_type"]: e["success"]
    for e in result.list_validation_results()[0]["results"]
})

# configuration.yml
data_source orders_dw:
  type: postgres
  host: dw-host
  port: 5432
  username: soda_monitor
  password: ${SODA_PG_PASSWORD}
  database: warehouse
  schema: ods

# checks/ods_orders.yml
checks for ods_orders:
  # 时效性：数据必须每 5 分钟内更新
  - freshness using column created_at < 5m
  # 完整性：行数不能低于下限
  - row_count > 100000
  - missing_count(order_id) = 0
  - missing_count(user_id) < 100
  # 唯一性：重复数量为零
  - duplicate_count(order_id) = 0
  # 有效性：金额取值区间
  - max(amount) < 1000000
  - min(amount) >= 0
  # Schema：不允许出现未声明的列
  - schema:
      warn:
        when unknown columns: 0

#!/bin/bash
# run_soda.sh
export SODA_PG_PASSWORD=$(aws secretsmanager get-secret-value \
  --secret-id dw/monitor --query SecretString --output text)

soda scan -d orders_dw -c configuration.yml checks/ods_orders.yml -s scan_results.json

# 退出码：0=通过，非 0=失败，可直接用于 CI 门禁
echo "Soda scan exit code: $?"

{
  "rule_id": "dq-rule-orders-0001",
  "expectation_type": "expect_column_values_to_not_be_null",
  "column": "order_id",
  "tier": "critical",
  "status": "active",
  "owner": "order-squad",
  "introduced_in": "dq-repo@v1.4.0",
  "alert_channels": ["#data-oncall", "email:dw@example.com"],
  "mttr_reference": "runbook://data-quality/null-order-id"
}

# .github/workflows/dq-gate.yml
name: Data Quality Gate
on:
  push:
    branches: [main]
    paths: ["models/**", "suites/**"]
  schedule:
    - cron: "0 * * * *"

jobs:
  run-gx-checkpoint:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: "3.11"
      - name: Install deps
        run: pip install "great_expectations>=1.0" snowflake-connector-python
      - name: Run validation
        env:
          SNOWFLAKE_PASSWORD: ${{ secrets.SNOWFLAKE_PASSWORD }}
        run: |
          python -m great_expectations checkpoint run ods_orders.daily_check
          # 失败则非零退出码阻断 CI
      - name: Upload data docs
        if: always()
        uses: actions/upload-artifact@v4
        with:
          name: data-docs
          path: great_expectations/uncommitted/data_docs/

# cardinality_anomaly.py
from collections import deque

class CardinalityDetector:
    """基于历史基数的 EWMA 基线检测"""
    def __init__(self, window: int = 28, z_threshold: float = 4.0):
        self.history: deque[float] = deque(maxlen=window)
        self.z_threshold = z_threshold

    def observe(self, distinct_count: float) -> dict:
        if len(self.history) < 10:
            self.history.append(distinct_count)
            return {"anomaly": False, "reason": "cold_start"}
        # EWMA 均值与波动
        alpha = 0.3
        mu, var = self.history[0], 0.0
        for x in self.history:
            mu = alpha * x + (1 - alpha) * mu
        var = sum((x - mu) ** 2 for x in self.history) / len(self.history)
        sigma = var ** 0.5 or 1.0
        z = abs(distinct_count - mu) / sigma
        self.history.append(distinct_count)
        return {
            "anomaly": z > self.z_threshold,
            "z_score": round(z, 2),
            "baseline_mean": round(mu, 2),
            "current": distinct_count,
        }

detector = CardinalityDetector()
for d in [1_200_000, 1_210_000, 1_195_000, 1_205_000, 1_198_000, 980_000]:
    print(d, detector.observe(d))

# mean_drift.py
import great_expectations as gx

context = gx.get_context(mode="file")
suite = context.add_expectation_suite("ods_orders.mean_drift")

# 基于近 28 天基线计算出的置信区间
suite.expectation_context.add_expectation(
    expect_column_mean_to_be_between(column="amount", min_value=118.0, max_value=132.0)
)
suite.expectation_context.add_expectation(
    expect_column_stdev_to_be_between(column="amount", min_value=50.0, max_value=70.0)
)

# quality_score.py
def compute_quality_score(checks: dict) -> dict:
    weights = {
        "completeness": 0.25,
        "uniqueness": 0.20,
        "accuracy": 0.25,
        "timeliness": 0.15,
        "schema": 0.15,
    }
    score = 0.0
    detail = {}
    for name, w in weights.items():
        ok, total = checks[name]["passed"], checks[name]["total"]
        rate = ok / total if total else 1.0
        detail[name] = round(rate, 3)
        score += w * rate
    grade = "A" if score >= 0.98 else "B" if score >= 0.95 else "C" if score >= 0.9 else "D"
    return {"score": round(score * 100, 1), "grade": grade, "detail": detail}

print(compute_quality_score({
    "completeness": {"passed": 8, "total": 9},
    "uniqueness": {"passed": 4, "total": 4},
    "accuracy": {"passed": 6, "total": 7},
    "timeliness": {"passed": 2, "total": 3},
    "schema": {"passed": 1, "total": 1},
}))

# openlineage_emit.yaml
job:
  namespace: dq
  name: ods_orders.daily_check
  facets:
    qualityCheck:
      producer: great_expectations/1.0
      score: 87.5
      grade: B
      failed_expectations:
        - expect_column_values_to_be_in_set.status
        - freshness
run:
  status: FAILED
  failed_at: "2026-09-27T02:00:00Z"
outputs:
  - namespace: postgres.warehouse
    name: ods.ods_orders
    facets:
      dataQuality:
        pass_rate: 0.875