4. 流式数据契约：Kafka Topic 的契约定义、消息格式约束、流处理中的 Schema 校验

好，我们进入流式数据契约的核心战场——Kafka。说实话，在流批一体落地过程中，Kafka Topic 往往是数据契约最容易「裸奔」的地方。批处理好歹还有个 Hive 表结构、Parquet Schema 可以约束，但 Kafka 消息？很多人直接塞 JSON，连个校验都没有。

我见过最夸张的一个项目，生产端改了字段类型，消费端直接反序列化崩溃，整个实时链路断了 40 分钟。为什么？因为没有契约。今天我们就来把这块短板补上。

4.1 Kafka Topic 的契约定义：不止是 Topic 名

很多人觉得 Kafka Topic 的契约就是定个名字、配个分区数。其实远远不够。我个人习惯把 Topic 契约拆成三个层次：

层次	内容	典型问题
元信息层	Topic 名称、分区数、副本数、保留策略、清理策略	命名混乱、保留时间不一致导致数据丢失
Schema 层	消息 Key/Value 的 Schema 定义、字段类型、默认值	字段类型变更不通知、消费端解析失败
语义层	消息的投递语义（at-least-once / exactly-once）、排序保证、分区键规则	乱序消费、重复消息处理逻辑缺失

你想想看，如果只定义 Topic 名，那跟没定义差不多。真正的契约要能回答三个问题：谁写的？写什么格式？怎么保证消费端能读懂？

4.2 消息格式约束：Avro、Protobuf 还是 JSON Schema？

说到消息格式，我经常被问到：「用 JSON 不行吗？轻量又灵活。」嗯，这里要注意——JSON 的灵活恰恰是它的致命伤。没有 Schema 约束的 JSON，字段名可以随意改，类型可以随意变，消费端根本没法做防御性编程。

我在项目中主要用三种格式，各有适用场景：

• Avro：流批一体场景的首选。天生带 Schema，支持 Schema 演化（向前/向后兼容），序列化后体积小。配合 Confluent Schema Registry 使用，生产端和消费端自动协商 Schema。
• Protobuf：如果你团队有 gRPC 背景，或者对性能要求极高，Protobuf 更合适。但要注意，Protobuf 的 Schema 演化规则比 Avro 严格，字段删除必须用 reserved 关键字。
• JSON Schema：适合外部系统对接、日志采集等低延迟敏感场景。但 JSON 序列化体积大，解析开销高，流批一体中慎用。

我个人的建议是：内部流转用 Avro，外部对接用 JSON Schema + 强制校验。曾经有个项目，业务方坚持用 JSON，结果上线后消费端频繁报错，最后花了三天排查，发现是某个字段从 int 变成了 string。从那以后，我对无 Schema 的消息格式有了心理阴影。

4.3 流处理中的 Schema 校验：从生产端到消费端

Schema 校验不是只在生产端做一次就完事了。流处理链路长，中间可能有多个消费者、多个处理节点。我把它分为三个环节：

1. 生产端校验：消息写入 Topic 前，必须通过 Schema Registry 的兼容性检查。不兼容的消息直接拒绝写入。
2. 消费端校验：消费者拉取消息时，自动从 Schema Registry 获取对应的 Schema，反序列化时做类型校验。如果 Schema 不匹配，可以降级或告警。
3. 流处理节点校验：在 Flink/Spark Streaming 等流处理引擎中，每个算子都应该做 Schema 校验。尤其是做 join、union 操作时，两边的 Schema 必须对齐。

为什么会这样？因为流处理是长周期运行的，Schema 可能在运行期间发生演化。如果不做校验，一个 Schema 变更就能让整个作业挂掉。

4.4 Schema Registry 的实战配置

说到 Schema Registry，我简单说一下核心配置。以 Confluent Schema Registry 为例，兼容性策略有几种：

策略	说明	适用场景
BACKWARD	新 Schema 可以读取旧 Schema 写入的数据	消费者先升级，生产者后升级
FORWARD	旧 Schema 可以读取新 Schema 写入的数据	生产者先升级，消费者后升级
FULL	同时满足 BACKWARD 和 FORWARD	生产者和消费者同时升级，但不确定顺序
NONE	不校验兼容性	开发测试环境，生产环境不建议使用

我个人习惯在生产环境用 FULL 策略。虽然限制多一些，但能最大程度避免 Schema 演化导致的线上事故。曾经有个团队用 BACKWARD，结果生产者先升级了 Schema，消费者还没升级，直接消费失败。FULL 策略至少能保证双向兼容。

下面是一个 Avro Schema 定义的示例：

{
  "type": "record",
  "name": "OrderEvent",
  "namespace": "com.example.contract",
  "fields": [
    {"name": "order_id", "type": "string"},
    {"name": "user_id", "type": "string"},
    {"name": "amount", "type": "double"},
    {"name": "status", "type": {"type": "enum", "name": "OrderStatus",
      "symbols": ["CREATED", "PAID", "SHIPPED", "CANCELLED"]}},
    {"name": "created_at", "type": {"type": "long", "logicalType": "timestamp-millis"}}
  ]
}

注意这里用了枚举类型和逻辑类型。枚举能约束字段的取值范围，逻辑类型能明确时间戳的精度。这些都是契约的一部分。

4.5 流批一体中的 Schema 对齐

流批一体最大的痛点之一，就是流式 Schema 和批式 Schema 不一致。Kafka Topic 的 Avro Schema 和 Hive 表的 Parquet Schema 可能定义不同。我见过一个案例：流处理用 Avro 的 decimal 类型，批处理用 Parquet 的 decimal 类型，精度定义不一致，导致数据对不上。

解决思路其实不复杂：统一 Schema 定义源。把 Schema Registry 作为唯一的 Schema 管理中心，Hive、Flink、Spark 都从 Schema Registry 拉取 Schema。这样流和批用的是同一份契约。

下面是一个 Flink SQL 的示例：

CREATE TABLE order_events (
  order_id STRING,
  user_id STRING,
  amount DOUBLE,
  status STRING,
  created_at TIMESTAMP(3)
) WITH (
  'connector' = 'kafka',
  'topic' = 'order-events',
  'properties.bootstrap.servers' = 'kafka:9092',
  'format' = 'avro-confluent',
  'avro-confluent.url' = 'http://schema-registry:8081',
  'avro-confluent.subject' = 'order-events-value'
);

你看，这样定义的表，Schema 完全由 Schema Registry 管理。如果生产端改了 Schema，Flink 会自动感知并做兼容性校验。不需要手动改 DDL，也不需要重启作业。

4.6 流式数据契约的完整流程

最后，我用一张图来总结流式数据契约的完整流程。这张图我画了很多次，每次给团队培训都会拿出来讲。

从这张图可以看到，Schema Registry 是整个流式数据契约的核心枢纽。生产端写入前先注册 Schema，消费端读取时自动获取 Schema，流批引擎都从同一个地方拉取 Schema 定义。这样就能保证整个链路的数据契约一致性。

好了，流式数据契约的核心内容就这些。说白了，就是三个要点：Topic 契约要完整、消息格式要约束、Schema 校验要贯穿全链路。做到这三点，你的实时数据管道就不会因为 Schema 变更而频繁断流。