第4章：实时数据源接入——Kafka、Binlog与Schema Registry
实时数仓的根基在哪？说白了，就在数据源接入这一层。你维度建模设计得再漂亮，如果数据进不来、进不准、进不快，一切都是空谈。这一章，我重点聊聊实时场景下最主流的三种接入方式：Kafka作为统一数据总线、Canal/Maxwell采集MySQL Binlog、以及日志采集配合Schema Registry。这些是我在多个项目中反复打磨过的套路，希望能帮你少踩几个坑。
4.1 Kafka：实时数据总线
Kafka在实时数仓里的地位，就像HDFS在离线数仓里一样——它是数据的“中枢神经”。我个人习惯把Kafka当作所有实时数据的统一入口，不管是业务库的Binlog、前端的埋点日志、还是微服务的调用链数据，统统先进Kafka，再由下游的Flink或Spark Streaming消费。
核心原则：实时数仓中，Kafka Topic的设计应尽量贴近业务主题（如订单、支付、用户行为），而不是直接映射数据库表。这样有利于后续维度建模的星型模型构建。
4.1.1 Topic分区策略
分区数怎么定？我见过不少团队拍脑袋设成10个或20个。其实有个简单的估算公式：
分区数 = max(目标吞吐量 / 单分区吞吐量, 消费者并行度)举个例子，假设你的业务峰值每秒要写入100MB数据，单分区能扛10MB/s，消费者并行度是8，那分区数至少取10。但要注意，分区太多会带来元数据压力和文件句柄开销。我在项目中遇到过分区数设到200+，结果Kafka Broker频繁Full GC的惨案……后来控制在50以内，稳得很。
| 场景 | 推荐分区数 | 说明 |
|---|---|---|
| 低吞吐（< 10MB/s） | 3 ~ 6 | 小规模业务，单机部署即可 |
| 中吞吐（10 ~ 50MB/s） | 6 ~ 24 | 常见电商、金融场景 |
| 高吞吐（> 50MB/s） | 24 ~ 50 | 大促、直播、IoT场景 |
4.1.2 消息格式与Key选择
消息体我建议统一用Avro或Protobuf，别用纯JSON。为什么？JSON没有Schema约束，字段类型变了、字段名改了，下游消费端直接挂掉。我曾经因为上游一个字段从int改成string，导致整个实时ETL链路中断了半小时……
Key的选择也很关键。如果你需要保证同一订单ID的所有变更按顺序消费，Key就用订单ID。如果不需要严格顺序，用随机Key来打散负载。
小技巧：在Kafka消息头里带上时间戳和数据源标识，方便后续做数据回溯和链路追踪。我习惯在每条消息的Header里塞一个traceId，排查问题的时候能省一半时间。
4.2 Canal/Maxwell：MySQL Binlog采集
实时数仓里最核心的数据源，往往是业务数据库。怎么把MySQL的变更实时同步到Kafka？主流方案就是Canal和Maxwell。这两个工具我都深度用过，说说我的感受。
4.2.1 选型对比
| 特性 | Canal | Maxwell |
|---|---|---|
| 开发语言 | Java | Java |
| 部署方式 | 独立服务 + 管理端 | 独立进程 |
| 支持DDL | 支持 | 支持 |
| 输出格式 | 自定义protobuf/JSON | JSON |
| HA机制 | ZooKeeper协调 | 内置 |
| 社区活跃度 | 高（阿里开源） | 中 |
我个人更倾向Canal，因为它的管理界面和HA机制更成熟。但如果你团队小、不想引入ZooKeeper，Maxwell的轻量部署也是个好选择。
4.2.2 核心配置示例
以Canal为例，一个典型的配置长这样：
# canal.properties
canal.serverMode = kafka
canal.mq.servers = kafka-broker1:9092,kafka-broker2:9092
canal.mq.topic = canal_binlog
canal.mq.partitionsNum = 6
canal.mq.partitionHash = order_id
# instance.properties
canal.instance.master.address = mysql-host:3306
canal.instance.dbUsername = canal_user
canal.instance.dbPassword = canal_pass
canal.instance.filter.regex = mydb\\.(orders|payments|users)
canal.instance.parser.parallel = true避坑指南：我曾经在生产环境遇到Canal消费延迟越来越大，排查了半天发现是MySQL的binlog_format没设成ROW。如果格式是STATEMENT或MIXED，Canal解析出来的数据会丢失字段级别的变更信息。记住：实时数仓场景下，binlog_format必须为ROW。
4.2.3 数据格式与维度建模的映射
Canal输出的每条消息包含before和after两个字段。在维度建模里，这对应着缓慢变化维（SCD）的Type 2处理。比如用户修改了手机号，before是旧号，after是新号。我一般会在Flink里根据optype（INSERT/UPDATE/DELETE）来维护维表的历史版本。
{
"database": "mydb",
"table": "users",
"type": "UPDATE",
"ts": 1700000000000,
"data": {
"id": 1001,
"name": "张三",
"phone": "13800138001"
},
"old": {
"phone": "13900139000"
}
}4.3 日志采集与Schema Registry
除了Binlog，日志也是实时数仓的重要数据源。前端埋点、服务端日志、IoT设备上报……这些数据量大、格式多样，怎么统一管理？我的答案是：Schema Registry + Avro。
4.3.1 为什么需要Schema Registry
你想想看，如果每个日志生产者都自己定义JSON格式，字段名随意、类型随意，下游消费端怎么解析？今天加个字段，明天改个类型，整个链路就乱了。Schema Registry就是解决这个问题的——它像一个“数据字典中心”，所有Schema统一注册、版本管理、兼容性校验。
核心价值：Schema Registry保证了生产者和消费者之间的契约一致性。生产者必须按注册的Schema发送数据，消费者可以按Schema自动解析，无需手动维护解析逻辑。
4.3.2 主流方案对比
| 方案 | 序列化格式 | Schema Registry | 适用场景 |
|---|---|---|---|
| Confluent Avro | Avro | Confluent Schema Registry | Kafka生态，最成熟 |
| Protobuf | Protobuf | Confluent / 自建 | 高性能、跨语言 |
| Thrift | Thrift | 自建 | 历史遗留系统 |
我个人推荐Confluent Avro方案，因为它和Kafka集成得最好，Flink、Spark都有原生支持。而且Confluent Schema Registry提供了REST API，方便做自动化管理。
4.3.3 日志采集链路实战
一个典型的日志采集链路是这样的：
- 应用埋点：在代码里用log4j/logback输出结构化日志，字段固定为JSON格式
- Filebeat/Flume采集：从日志文件里读取数据，发送到Kafka
- Schema校验：在Kafka Connect或Flink里，用Schema Registry校验消息格式
- 写入Kafka Topic：校验通过的数据写入对应的Topic，失败的数据进入死信队列
经验之谈：我建议在日志采集阶段就做Schema校验，而不是等到ETL阶段。早发现、早隔离，避免脏数据污染下游。我曾经在凌晨大促时，因为一个埋点字段类型错误，导致整个实时报表链路瘫痪……从那以后，我强制所有日志Topic都配了Schema Registry校验。
4.4 本章知识体系总览
下面这张图，我把本章的核心知识点串起来了。你可以把它当作实时数据源接入的“作战地图”。
从这张图可以清楚看到：数据从MySQL、日志、其他源进来，经过Canal/Filebeat等工具采集，再通过Schema Registry做格式统一，最终汇入Kafka总线。下游的Flink/Spark消费后，才能进行维度建模的落地。每一步都有坑，每一步也都有最佳实践。
最后提醒一句：实时数据源接入不是“配通就行”的事。你需要在接入阶段就考虑好Schema兼容性、分区策略、死信处理、监控告警。否则，等数据量上来再改，代价会非常大。我见过太多团队在接入阶段图省事，结果后期花了几倍的时间来填坑。