DataPipeline

数据管道是一组可重复执行的数据处理过程，负责把数据从源系统可靠地传递到分析、检索或机器学习系统。合格的数据管道不仅能够搬运数据，还应具备质量校验、失败重试、血缘追踪、幂等写入和运行监控能力。

基本结构

flowchart LR
    A[业务数据库] --> D[采集层]
    B[日志与事件] --> D
    C[外部接口] --> D
    D --> E[原始数据层]
    E --> F[清洗与标准化]
    F --> G[质量校验]
    G --> H[主题数据层]
    H --> I[数据仓库]
    H --> J[特征平台]
    H --> K[检索与报表]
    G -->|异常数据| L[隔离区]

批处理与流处理

批处理按固定时间窗口读取有界数据，适合日报、历史回算和计算成本敏感的任务。流处理持续消费事件，适合监控、推荐和实时风控。架构选择应由业务时效要求决定，不应仅以技术新旧作为判断依据。

维度	批处理	流处理
数据边界	有界	通常无界
延迟	分钟至小时	毫秒至分钟
重算	相对简单	需要回放机制
状态管理	较少	窗口与检查点较复杂

设计原则

幂等性：相同输入重复执行后，目标状态保持一致。
可恢复性：任务失败后能够从检查点继续，或安全地重新执行。
数据契约：明确字段类型、可空性、唯一性和兼容策略。
可观测性：记录吞吐量、延迟、失败率、数据新鲜度和质量指标。
数据血缘：能够定位一个输出字段来自哪个源字段以及经历了哪些转换。

简化实现

from dataclasses import dataclass
from datetime import datetime

@dataclass(frozen=True)
class Event:
    event_id: str
    user_id: str
    occurred_at: datetime

def transform(row: dict) -> Event:
    return Event(
        event_id=str(row["event_id"]),
        user_id=str(row["user_id"]),
        occurred_at=datetime.fromisoformat(row["occurred_at"]),
    )

def validate(event: Event) -> None:
    if not event.event_id or not event.user_id:
        raise ValueError("event_id and user_id are required")

def run_pipeline(source, sink) -> None:
    for row in source.read():
        event = transform(row)
        validate(event)
        sink.upsert(event.event_id, event)