数据管道基础：从采集到入库

做 AI 项目时，模型只是最后一步，前面往往要先把数据跑通。数据管道（Data Pipeline）就是一条把原始数据变成可用数据的流水线：从源头采过来、洗干净、转成统一格式、再存进数据系统。管道稳不稳，直接决定后面分析和训练的上限。

一条管道通常分几段

最朴素的分法是四段。采集（Ingestion）负责从数据库、日志、接口、文件里把数据拉进来；清洗（Cleaning）处理缺失、去重、纠错；转换（Transformation）做格式统一、特征派生、归一化；入库（Load）把结果写进数据仓库、特征库或向量库。每段都可能独立失败，所以最好分段可重试。

def run_pipeline(raw):
    df = ingest(raw)          # 采集
    df = clean(df)            # 去空、去重、纠错
    df = transform(df)        # 统一字段与格式
    load(df)                  # 写入目标库
    return df

批处理还是流式

数据量小、时效要求低，用批处理（Batch）按小时或天跑就够了，简单好查。数据要实时（监控、推荐、风控），就用流式（Streaming），数据一来就处理。两者不是非此即彼，很多管道是「批处理兜底 + 流式增量」的组合：流式保新鲜，批处理保完整。

几个容易踩的坑

最常见的是 schema 漂移：上游改了字段，下游没跟上，整条管道报错。解决办法是给数据加校验（类型、范围、非空）并尽早失败告警。还要注意幂等——同一份数据跑两次不能重复入库，给每条记录稳定的唯一键是关键。最后留好可观测性：每段花多久、失败多少、数据量多大，没有这些排障只能靠猜。

小结

数据管道把零散原始数据变成干净可用的数据，是 AI 项目真正的基础设施。把它拆成采集、清洗、转换、入库几段，按批处理或流式选型，并对 schema、幂等和可观测性提前设防，管道才能长期稳定运行，让下游分析和训练少踩坑。

参考与延伸阅读

• 各云厂商的数据管道与 ETL 文档（如 Airflow、Spark、Kafka 生态），讲解调度、批流与连接器选型。已核验。
• 具体组件搭配以你的数据规模与时效要求为准，不同团队对「实时」的阈值差异很大。待核实。