📚 课程目录第4章 Pipeline组件解构

Pipeline 组件解构：四个核心积木

你有没有想过，一个完整的元数据 Pipeline 到底由哪些东西拼起来的？

我刚开始搭这类系统时，也犯过糊涂。以为写几个脚本就能搞定。后来踩的坑多了，才发现一个靠谱的 Pipeline，离不开四个核心组件：数据源、转换层、存储层、调度器。

说白了，它们就像一辆车的四个轮子。少一个，你跑不远。配合不好，你跑不稳。

1. 数据源：你的原材料在哪

数据源，就是 Pipeline 的起点。你想想看，没有原材料，后面再牛的生产线也是白搭。

我见过的元数据源，五花八门。最常见的有几种：

数据库：MySQL、PostgreSQL、MongoDB，里面存着表结构、字段注释、索引信息。
消息队列：Kafka、RocketMQ，实时流过来的 Schema 变更。
代码仓库：Git 里的配置文件、DDL 脚本、YAML 定义。
云服务 API：AWS Glue、阿里云 DataWorks 的元数据接口。

我曾经掉过的坑：把数据源直接绑死在某种数据库上。后来业务换库，整个 Pipeline 重写。我建议你在设计阶段就给数据源加一层抽象接口。这样换个源，只改适配器，不动核心逻辑。

2. 转换层：数据的炼油厂

元数据从源头拿来，通常是"生"的。比如一个 DDL 语句，你需要解析出表名、字段类型、约束条件。这个解析和清洗的过程，就是转换层干的活。

我个人习惯把转换层拆成三步：

解析：把原始格式转成结构化数据。比如 SQL 解析成 AST。
清洗：去重、补缺、标准化。字段名统一驼峰还是下划线，在这里定。
丰富：打标签、算血缘，或者关联业务元数据。

实际案例：我在做个数据血缘项目时，解析一条 SQL 只需要 2 毫秒。但清洗和关联花了大头。所以，转换层往往是 Pipeline 的性能瓶颈。写代码时，这一块值得多花心思优化。

转换层的输出呢？通常是一个统一的元数据模型。我推荐用 Protobuf 或者 Avro 来定义这个模型，因为跨语言兼容性好。

syntax = "proto3";

message ColumnMetadata {
  string name = 1;
  string data_type = 2;
  bool is_nullable = 3;
  string comment = 4;
}

message TableMetadata {
  string database = 1;
  string table_name = 2;
  repeated ColumnMetadata columns = 3;
  int64 updated_at = 4;
}

3. 存储层：快照放哪

元数据转换完了，总得有个地方存着吧。存储层就是干这个的。但它不只是简单地放个文件那么简单。这里有学问。

你想想看，元数据要支持版本回滚、历史查询、快速检索。所以选存储方案时，我一般会关注几个点：

存储方案	适用场景	快照方式	我踩过的坑
MySQL/PostgreSQL	少量元数据，用 JSON 字段存快照	历史表 + 时间戳	快照膨胀太快，我后来加了 TTL
HDFS/对象存储	海量元数据，Parquet 格式存	目录分区 + 增量文件	小文件太多，NameNode 抗议了
Elasticsearch	需要全文检索、快速查询	索引别名 + 版本号	——
专用元数据平台	企业级数据治理	内置快照机制	定制成本高，用过一次就放弃了

说个经验：如果你的快照需要频繁回溯，我建议用不可变存储模式——每次快照都写成新文件，不覆盖旧的。虽然占空间，但恢复时不用纠结。

4. 调度器：谁来决定什么时候干活

数据源准备好了，转换层写好了，存储层搭好了。然后呢？总得有个人喊"开始干活"吧。调度器就是这个角色。

调度器不负责实际数据处理，只负责触发和管理任务。它的理解，我总结成三个核心职责：

定时触发：比如每天凌晨 2 点执行一次全量快照。
事件驱动：Kafka 来了新消息，立马触发增量处理。
失败重试与告警：任务挂了，自动重试 3 次，还不行就发钉钉/邮件。

我早期用的是 Crontab，简单粗暴。后来任务多了，依赖关系复杂了，Crontab 就扛不住了。现在主流的选择有：

Apache Airflow：DAG 编排，生态成熟。很适合复杂的元数据 Pipeline。
DolphinScheduler：国产的，可视化界面不错，大数据圈用得多。
Prefect：轻量级，Python 亲儿子。我小团队时特别喜欢它。

嗯，这里要注意：调度器最怕"死锁"或"资源饿死"。我做过一个 Pipeline，全量任务和增量任务抢同一个表的写锁，结果两边都卡死。后来我强制为全量和增量分配不同的资源池，才解决问题。

一个典型的调度配置长什么样？以 Airflow 为例，大概是这样：

from airflow import DAG
from datetime import datetime

with DAG(
    dag_id='metadata_snapshot_pipeline',
    schedule_interval='0 2 * * *',  # 每天凌晨2点
    start_date=datetime(2023, 1, 1),
    catchup=False
) as dag:
    # 任务1: 采集元数据
    ingest_task = IngestOperator(task_id='ingest_metadata')
    # 任务2: 转换清洗
    transform_task = TransformOperator(task_id='transform_metadata')
    # 任务3: 生成快照并存储
    store_task = StoreOperator(task_id='store_snapshot')

    ingest_task >> transform_task >> store_task

组件之间如何协作

讲完四个组件，你可能想问：它们之间到底怎么配合的？

用大白话说就是：调度器告诉数据源"该起床了"，数据源把原生数据吐给转换层，转换层洗完后交给存储层。存储层生成一个带时间戳的快照文件。调度器确认所有步骤都成功了，就记录一下本次快照的元数据。

如果中间某一步失败了，调度器会尝试重试。重试还不行，就发告警，然后留一个"脏数据"标记，方便人工介入排查。

个人习惯：我喜欢在每个快照文件里加一个 _pipeline_status.json，记录本次执行的每个组件版本号、耗时、异常信息。这样排查问题时，不用翻半天日志，看一眼 JSON 就知道问题出在哪。

这四个组件，说复杂也复杂，说简单也简单。但只要你把它们的角色和边界理清了，你的元数据 Pipeline 就能做到可重复、可恢复、可追溯。这也是我们这门课的核心目标。