业务系统里的数据架构：不只是画 ER 图

data-arch-demo/
├── config/
│   ├── sync_pipeline.yaml      # 数据同步配置
│   └── storage_routing.yaml    # 存储路由规则
├── models/
│   ├── operational.py          # 操作层模型定义
│   ├── analytical.py           # 分析层模型定义
│   └── archival.py             # 影像层模型定义
├── pipelines/
│   ├── op_to_analytical.py     # 操作层 → 分析层同步
│   ├── op_to_archival.py       # 操作层 → 影像层归档
│   └── quality_check.py        # 数据质量校验
└── main.py

# config/storage_routing.yaml
# 定义不同数据域的存储路由规则
routing:
  order:
    operational:
      engine: postgresql
      database: biz_ops
      consistency: strong
      retention: 90d          # 操作层只保留近期数据
    analytical:
      engine: clickhouse
      database: biz_analytics
      consistency: eventual
      sync_mode: realtime     # 实时同步，延迟 < 5s
      retention: 365d
    archival:
      engine: s3_parquet      # 对象存储 + Parquet 格式
      bucket: data-archive
      partition: date         # 按日期分区
      sync_mode: daily_batch  # 每日批量归档
      retention: 7y

  user_profile:
    operational:
      engine: postgresql
      database: biz_ops
      consistency: strong
      retention: permanent    # 用户资料长期保留在操作层
    analytical:
      engine: clickhouse
      database: biz_analytics
      consistency: eventual
      sync_mode: hourly_batch
      retention: 365d
    archival:
      skip: true              # 用户资料不需要额外归档

# pipelines/op_to_analytical.py
"""操作层 → 分析层：近实时同步管道"""

import time
from datetime import datetime, timedelta
from sqlalchemy import create_engine, text
from clickhouse_driver import Client

# 从配置文件加载，这里用硬编码演示
PG_URL = "postgresql://user:pass@localhost:5432/biz_ops"
CH_URL = "http://localhost:8123"

pg_engine = create_engine(PG_URL)
ch_client = Client(host="localhost", port=9000, database="biz_analytics")

SYNC_WATERMARK = {}  # 记录每张表上次同步的位置


def sync_orders(batch_size=5000, lag_seconds=10):
    """将操作层订单增量同步到分析层"""

    # 用更新时间作为水位线，避免漏同步
    cutoff = datetime.utcnow() - timedelta(seconds=lag_seconds)
    last_sync = SYNC_WATERMARK.get("orders", datetime.utcnow() - timedelta(hours=1))

    if last_sync >= cutoff:
        print("[sync] 无新数据需要同步")
        return 0

    query = text("""
        SELECT order_id, user_id, amount, status, created_at, updated_at
        FROM orders
        WHERE updated_at > :last_sync AND updated_at <= :cutoff
        ORDER BY updated_at
        LIMIT :batch_size
    """)

    with pg_engine.connect() as conn:
        rows = conn.execute(query, {
            "last_sync": last_sync,
            "cutoff": cutoff,
            "batch_size": batch_size,
        }).fetchall()

    if not rows:
        SYNC_WATERMARK["orders"] = cutoff
        return 0

    # 写入 ClickHouse
    ch_client.execute(
        "INSERT INTO orders_analytical VALUES",
        rows  # 直接传入 tuple 列表
    )

    # 推进水位线
    SYNC_WATERMARK["orders"] = cutoff
    print(f"[sync] 已同步 {len(rows)} 条订单，水位线推进到 {cutoff}")
    return len(rows)


if __name__ == "__main__":
    # 模拟持续同步循环
    while True:
        sync_orders()
        time.sleep(5)  # 每 5 秒轮询一次

-- ClickHouse 建表
CREATE TABLE IF NOT EXISTS orders_analytical (
    order_id    String,
    user_id     String,
    amount      Decimal64(2),
    status      String,
    created_at  DateTime,
    updated_at  DateTime
) ENGINE = MergeTree()
ORDER BY (created_at, order_id)
PARTITION BY toYYYYMM(created_at);

# pipelines/quality_check.py
"""每日数据一致性校验"""

from sqlalchemy import create_engine, text
from clickhouse_driver import Client

pg_engine = create_engine("postgresql://user:pass@localhost:5432/biz_ops")
ch_client = Client(host="localhost", port=9000, database="biz_analytics")


def check_orders_count(date_str: str):
    """校验某日订单行数是否一致"""

    # 操作层
    with pg_engine.connect() as conn:
        pg_count = conn.execute(text("""
            SELECT COUNT(*) FROM orders
            WHERE created_at::date = :date
        """), {"date": date_str}).scalar()

    # 分析层
    ch_count = ch_client.execute(
        "SELECT COUNT(*) FROM orders_analytical WHERE toDate(created_at) = :date",
        {"date": date_str}
    )[0][0]

    diff = pg_count - ch_count
    status = "✅ PASS" if diff == 0 else f"❌ FAIL (差 {diff} 条)"
    print(f"[校验] {date_str} 订单: PG={pg_count}, CH={ch_count} → {status}")
    return diff == 0


if __name__ == "__main__":
    from datetime import datetime, timedelta
    # 校验昨天的数据
    yesterday = (datetime.utcnow() - timedelta(days=1)).strftime("%Y-%m-%d")
    check_orders_count(yesterday)