📚 课程目录第3章流式数据资产建模

流式数据资产建模：Kafka Topic 元数据模型与 Schema Registry 集成

流式数据资产，说白了就是「一直在动的数据」。跟离线表不一样，它没有「存储」这个概念，只有「管道」和「事件」。我在做实时数仓那几年，最头疼的就是怎么把这种「流动的资产」管起来。

今天咱们就聊聊三个核心问题：Kafka Topic 的元数据该怎么建模、Schema Registry 怎么集成进来、以及流式血缘怎么追踪。嗯，这三个点搞定了，流式资产的管理框架基本就立住了。

Kafka Topic 元数据模型

先说说 Topic。很多人觉得 Topic 不就是个名字嘛，有什么好建模的？其实不然。你想想看，一个 Topic 在数据目录里应该长什么样？

我个人习惯把 Topic 元数据分成三层：

基础层：Topic 名称、分区数、副本数、创建时间、所属集群
配置层：保留策略（retention.ms）、压缩策略（cleanup.policy）、消息大小限制
运行时层：当前消息总量、最近写入速率、消费者 lag 情况

我在项目中遇到过一个问题：业务方说「这个 Topic 数据丢了」，结果查了半天，发现是 retention 设置成了 1 小时。从那以后，我要求所有 Topic 的配置信息必须进数据目录，而且要有变更历史。

下面是一个简化的元数据模型结构：

{
  "topic_name": "order_events",
  "cluster": "kafka-prod-01",
  "partitions": 12,
  "replication_factor": 3,
  "configs": {
    "retention.ms": "604800000",
    "cleanup.policy": "delete",
    "max.message.bytes": "1048576"
  },
  "stats": {
    "total_messages": 1250000000,
    "bytes_in_per_sec": 45.2,
    "bytes_out_per_sec": 38.7
  },
  "owners": ["data-platform-team"],
  "created_at": "2024-03-15T10:30:00Z"
}

小技巧：Topic 的「业务含义」字段一定要单独拎出来。很多团队只记技术属性，结果半年后谁都不知道这个 Topic 是干嘛的。我建议加一个 business_description 字段，强制填写。

Schema Registry 集成

Topic 的元数据只是「壳」，真正的核心是里面的数据长什么样。这就是 Schema Registry 的活儿了。

我记得第一次接触 Avro 的时候，觉得这东西真麻烦。后来线上出了几次「字段类型不匹配」的故障，我才意识到：没有 Schema 约束的流式数据，就是一场灾难。

集成 Schema Registry 到数据目录，我建议至少做三件事：

注册 Schema 版本：每次 Schema 变更都生成一个新版本，保留历史
兼容性检查：记录当前使用的兼容性策略（BACKWARD、FORWARD、FULL 等）
字段级映射：把 Schema 字段映射到数据目录的字段模型里

来看一个实际的 Schema 注册示例：

{
  "subject": "order_events-value",
  "version": 3,
  "schema": {
    "type": "record",
    "name": "OrderEvent",
    "fields": [
      {"name": "order_id", "type": "string"},
      {"name": "user_id", "type": "string"},
      {"name": "amount", "type": "double"},
      {"name": "status", "type": {"type": "enum", "symbols": ["CREATED", "PAID", "SHIPPED"]}},
      {"name": "event_time", "type": "long", "logicalType": "timestamp-millis"}
    ]
  },
  "compatibility": "BACKWARD",
  "id": 1024
}

避坑指南：我曾经遇到一个团队，把 Schema 版本号直接写死在消费者代码里。结果 Schema 升级后，老版本消费者全部挂掉。正确的做法是：消费者只订阅 Topic，不绑定版本号，让 Schema Registry 自动做兼容性转换。

流式血缘追踪

血缘追踪在离线场景下已经很成熟了，但流式场景完全是另一回事。为什么？因为流式数据没有「快照」，你没法像离线表那样做全量回溯。

我做过一个实时数仓项目，数据链路是这样的：

Kafka Topic A（原始订单事件）
→ Flink 作业（清洗、聚合）
→ Kafka Topic B（清洗后的订单事件）
→ Flink 作业（关联用户维度）
→ Kafka Topic C（宽表事件）
→ ClickHouse 表（实时写入）

这条链路里，任何一个环节出问题，下游的数据就全歪了。所以流式血缘的核心是：记录数据在管道中的每一次「变形」。

我建议用「节点-边」模型来建模流式血缘：

元素	说明	示例
源节点	数据产生的起点	Kafka Topic、数据库 CDC
处理节点	数据经过的计算逻辑	Flink SQL、Spark Streaming
目标节点	数据最终落地的位置	Kafka Topic、HBase、ClickHouse
边	数据流动的方向和转换规则	字段映射、过滤条件、聚合逻辑

下面这张图展示了流式血缘的核心结构：

  核心要点：流式血缘的难点不在于「画图」，而在于「自动采集」。我建议在 Flink 作业的算子层面埋点，把 input/output Topic、字段映射、过滤条件都记录下来，统一上报到数据目录的血缘服务。

实战：元数据采集与同步

理论说完了，咱们看看实际怎么落地。我一般用「定时采集 + 事件驱动」双模式：

定时采集：每 5 分钟扫描一次 Kafka 集群，发现新 Topic 自动注册到数据目录
事件驱动：监听 Schema Registry 的变更事件，Schema 版本更新时实时同步

这里给一个简单的采集脚本思路：

# 伪代码：Kafka Topic 元数据采集
def collect_topic_metadata(cluster, admin_client):
    topics = admin_client.list_topics()
    for topic in topics:
        # 获取分区信息
        partitions = admin_client.describe_topic(topic)
        # 获取配置
        configs = admin_client.describe_configs(topic)
        # 获取 Schema（如果存在）
        schema = schema_registry.get_latest_schema(topic + "-value")
        # 写入数据目录
        data_catalog.register_topic(
            cluster=cluster,
            topic_name=topic,
            partitions=len(partitions),
            configs=configs,
            schema=schema
        )

经验之谈：采集频率别太高，尤其是大集群。我见过有人设成 10 秒一次，结果 Kafka 集群的元数据请求把 Controller 打挂了。5 分钟一次足够了，实时性靠事件驱动来补。

最后说一句：流式资产建模的核心不是「管住数据」，而是「让数据可理解」。你想想看，一个 Topic 在数据目录里，如果能一眼看出它的 Schema、血缘、消费方、生产方，那这个目录就真正有价值了。