4. 数据接入与解析：Kafka Topic 设计规范、数据格式选择与 Schema Registry
数据接入是实时特征平台的入口。入口没设计好，后面全是坑。我见过不少团队，Topic 命名乱得像草稿纸，数据格式换来换去，线上出问题连字段含义都查不到。这一章，我把自己的实践经验拆开来讲。
4.1 Kafka Topic 设计规范
Topic 是 Kafka 里最基础的概念。但很多人把它当消息队列用，随便起个名字就上线了。嗯，这里要注意——Topic 设计直接影响到后续的运维、权限管理和数据治理。
4.1.1 命名规范
我个人习惯用三段式命名：
<数据域>.<数据主题>.<数据格式>举个例子：
user.behavior.click.avroorder.transaction.created.protobufrisk.fraud.decision.json
为什么这么设计？
- 数据域：区分业务线，比如 user、order、risk。权限控制可以直接按前缀匹配。
- 数据主题：描述具体事件，比如 behavior.click、transaction.created。一眼能看出是什么数据。
- 数据格式：标明序列化方式。这个很重要——消费者拿到 Topic 名就知道该用什么反序列化器。
test_log，上线半年后没人知道它是什么数据、谁在用。最后只能全量回溯代码，浪费了两天时间。命名规范不是形式主义，是救命稻草。
4.1.2 分区与副本策略
| 参数 | 推荐值 | 说明 |
|---|---|---|
| 分区数 | 3 ~ 12（按吞吐量估算） | 分区数 = 目标吞吐量 / 单分区吞吐量。单分区写吞吐约 10MB/s。 |
| 副本因子 | 3 | 至少 2，推荐 3。保证 Broker 宕机时数据不丢。 |
| min.insync.replicas | 2 | 配合 acks=all 使用，确保写入一致性。 |
| retention.ms | 72 小时（按业务需求） | 实时特征场景一般不需要长保留，节省磁盘。 |
分区数不是越多越好。分区太多会增加 Broker 的元数据压力，也会让消费者 Rebalance 变慢。我一般按峰值吞吐量的 1.5 倍预留分区。
4.2 数据格式选择：Avro / Protobuf / JSON
数据格式选型，说白了就是平衡三个东西：性能、兼容性、可读性。我直接给结论，再讲为什么。
4.2.1 三种格式对比
| 特性 | Avro | Protobuf | JSON |
|---|---|---|---|
| 序列化大小 | 小（二进制） | 小（二进制） | 大（文本） |
| 序列化/反序列化速度 | 快 | 快 | 慢 |
| Schema 定义 | JSON 格式，自带 IDL | .proto 文件 | 无（或手写校验） |
| Schema 演化 | 强（支持向前/向后兼容） | 强（需手动管理） | 弱（字段增减容易出错） |
| 可读性 | 差（需反序列化） | 差（需反序列化） | 好（肉眼可读） |
| 生态支持 | Kafka 原生支持 | 需额外配置 | 通用 |
4.2.2 我的选型建议
- Avro：实时特征平台的首选。Kafka 和 Schema Registry 原生集成，序列化性能好，Schema 演化能力强。我个人大部分场景都用 Avro。
- Protobuf：如果你团队已经重度使用 gRPC 或 Protobuf，可以统一。但要注意，Protobuf 在 Kafka 里需要自己处理 Schema 注册和反序列化，稍微麻烦一点。
- JSON：只适合调试、日志、或者数据量极小的场景。线上生产环境，JSON 的序列化开销和 Schema 管理成本太高。
4.3 Schema Registry 的使用
Schema Registry 是 Avro/Protobuf 的“身份证中心”。它负责存储、校验、分发 Schema。没有它，你根本没法做 Schema 演化。
4.3.1 为什么需要 Schema Registry？
你想想看，生产者升级了 Schema，加了一个字段。消费者如果还是老版本，反序列化就会报错。Schema Registry 通过兼容性检查，保证生产者和消费者之间的 Schema 是匹配的。
我遇到过最惨的一次：生产环境直接改了 Avro Schema，没经过 Registry，结果所有消费者全部崩溃。从那以后，我强制要求所有 Avro 数据必须经过 Schema Registry。
4.3.2 兼容性策略
| 策略 | 说明 | 适用场景 |
|---|---|---|
| BACKWARD | 新 Schema 可以读旧数据 | 默认策略，推荐使用 |
| FORWARD | 旧 Schema 可以读新数据 | 消费者先升级的场景 |
| FULL | 双向兼容 | 严格要求的场景 |
| NONE | 不检查 | 仅调试用，禁止生产 |
我个人习惯用 BACKWARD。因为大多数场景是生产者先升级，消费者后升级。只要保证新 Schema 能读旧数据，就不会出问题。
4.3.3 集成示例
下面是一个 Avro 生产者集成 Schema Registry 的代码片段：
// 生产者配置
Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "io.confluent.kafka.serializers.KafkaAvroSerializer");
props.put("schema.registry.url", "http://localhost:8081");
// 定义 Avro Schema
String userSchema = "{\"type\":\"record\",\"name\":\"User\",\"fields\":["
+ "{\"name\":\"id\",\"type\":\"int\"},"
+ "{\"name\":\"name\",\"type\":\"string\"},"
+ "{\"name\":\"age\",\"type\":\"int\"}"
+ "]}";
Schema.Parser parser = new Schema.Parser();
Schema schema = parser.parse(userSchema);
// 发送消息
try (Producer<String, GenericRecord> producer = new KafkaProducer<>(props)) {
GenericRecord record = new GenericData.Record(schema);
record.put("id", 1001);
record.put("name", "张三");
record.put("age", 28);
producer.send(new ProducerRecord<>("user.profile.avro", "key1", record));
}消费者端只需要配置相同的 schema.registry.url，反序列化时会自动拉取 Schema。
4.4 整体架构图
下面这张图展示了数据接入与解析的核心流程：
4.5 避坑指南
- Topic 不要用下划线开头：Kafka 内部 Topic 以下划线开头（如
__consumer_offsets），容易混淆。 - Schema 版本号不要手动改：Schema Registry 自动管理版本，手动改会导致兼容性检查失败。
- 不要在生产环境用 NONE 兼容策略：一旦 Schema 不兼容，所有消费者都会挂。我见过有人图省事设了 NONE，结果线上事故。
- 分区数一旦确定，不要轻易减少：Kafka 不支持减少分区数。如果预估不足，只能重建 Topic。