4. 数据采集层：实时数据源接入与日志采集
数据采集层是整个实时OLAP系统的「水源地」。你想想看，如果源头的水质不行，下游再怎么折腾也白搭。我这些年见过太多团队，花大把精力优化查询引擎，结果数据采集端丢数据、延迟高，最后整个系统都不可信。
这一章，咱们就聊聊实时数据源怎么接、日志怎么采、数据库变更怎么抓，以及数据格式怎么选。都是我在实战中踩过的坑，希望能帮你少走弯路。
4.1 消息队列选型：Kafka vs Pulsar
消息队列是实时数据采集的「高速公路」。选对了，后续扩展很顺畅；选错了，改起来成本极高。
核心观点：我个人习惯，如果团队Kafka经验丰富、业务场景以高吞吐日志采集为主，优先选Kafka。如果有多租户隔离、跨地域复制、队列模型等复杂需求，Pulsar更合适。
| 特性 | Kafka | Pulsar |
|---|---|---|
| 存储模型 | 分区日志，依赖磁盘顺序读写 | 分层存储：BookKeeper + 计算节点 |
| 消息模型 | 仅支持发布-订阅 | 支持发布-订阅 + 队列模型 |
| 多租户 | 通过Topic命名空间模拟 | 原生支持，资源隔离更彻底 |
| 延迟消息 | 需自行实现 | 内置支持 |
| 运维复杂度 | 中等，社区成熟 | 较高，组件多 |
| 吞吐量 | 极高（百万级/秒） | 高（十万级/秒） |
我在项目中遇到过一个问题：用Kafka做日志采集，业务量突然翻倍，分区数不够导致消费端积压。后来我们提前做了分区预评估，按峰值流量的1.5倍预留分区，才稳住了。
避坑指南：Kafka的分区数一旦确定，后续扩容需要重新分配数据，成本很高。建议初始分区数 = 预期峰值分区数 × 1.5，并开启自动Leader再平衡。
4.2 日志采集方案：Filebeat vs Fluentd
日志采集是实时数据源接入的「最后一公里」。选Filebeat还是Fluentd？我直接说结论：
- Filebeat：轻量、简单、资源占用低。适合「采集→发送」这种单一场景。
- Fluentd：插件丰富、路由灵活。适合需要复杂数据处理的场景。
我曾经在一个金融项目中，用Filebeat采集应用日志，结果发现日志格式不统一，下游解析很痛苦。后来我们在Filebeat前面加了一层Logstash做格式标准化，虽然多了一个组件，但整体稳定性提升不少。
注意：Filebeat默认的背压机制比较弱。如果Kafka写入变慢，Filebeat会不断重试，可能导致内存飙升。建议配置 queue.mem.events 限制内存队列大小。
# Filebeat 配置示例
filebeat.inputs:
- type: log
paths:
- /var/log/app/*.log
fields:
app: my-service
env: production
output.kafka:
hosts: ["kafka1:9092", "kafka2:9092"]
topic: "app-logs"
compression: gzip
max_message_bytes: 1048576
# 内存队列限制
queue.mem:
events: 4096
flush.min_events: 512
flush.timeout: 1s
4.3 CDC技术：Debezium vs Canal
CDC（Change Data Capture）是实时数据采集的「核武器」。它能捕捉数据库的每一次变更，然后实时同步到消息队列。说白了，就是让数据库也能「说话」。
我对比一下两个主流方案：
| 特性 | Debezium | Canal |
|---|---|---|
| 支持数据库 | MySQL、PostgreSQL、MongoDB、Oracle等 | 仅MySQL（阿里系） |
| 部署方式 | Kafka Connect插件 | 独立服务 |
| 数据格式 | Avro/JSON（Schema Registry） | JSON/Protobuf |
| 社区活跃度 | 高（Red Hat维护） | 中（阿里开源） |
| 运维复杂度 | 依赖Kafka Connect | 独立部署，相对简单 |
我记得有一次，用Debezium同步MySQL数据到Kafka，结果发现Binlog格式没设置成ROW模式，导致数据变更捕获不全。嗯，这里要注意：Debezium要求MySQL的Binlog格式必须是ROW，且开启GTID。
关键配置：
-- MySQL 必须开启
binlog_format = ROW
binlog_row_image = FULL
gtid_mode = ON
enforce_gtid_consistency = ON
4.4 数据格式与序列化：Avro vs Protobuf
数据格式选得好，后续处理省一半的力。我见过太多团队用JSON做实时数据格式，结果序列化/反序列化开销大，Schema变更还容易出问题。
我个人推荐：Avro 用于日志采集场景，Protobuf 用于CDC场景。为什么？
- Avro：天生支持Schema演化，配合Schema Registry，可以做到向前/向后兼容。日志字段经常增减，Avro很合适。
- Protobuf：序列化后体积更小，解析速度更快。CDC数据字段相对稳定，Protobuf更高效。
实战经验：我在一个电商项目中，用Avro做日志采集的序列化格式。刚开始觉得多维护一个Schema Registry很麻烦，但后来业务方频繁加字段，Avro的兼容性帮了大忙。如果当时用JSON，每次加字段都要改下游所有消费端。
// Avro Schema 示例
{
"type": "record",
"name": "OrderEvent",
"namespace": "com.example",
"fields": [
{"name": "order_id", "type": "string"},
{"name": "user_id", "type": "int"},
{"name": "amount", "type": "double"},
{"name": "status", "type": {"type": "enum", "name": "OrderStatus", "symbols": ["CREATED", "PAID", "SHIPPED"]}},
{"name": "created_at", "type": "long", "logicalType": "timestamp-millis"}
]
}
4.5 整体架构图
下面这张图，是我在实际项目中用过的数据采集层架构。你可以看到，从日志、数据库、业务API三个源头，经过采集、传输、序列化，最终进入消息队列，供下游实时计算引擎消费。
4.6 避坑总结
最后，我把这些年踩过的坑总结一下：
- Kafka分区数别拍脑袋：先做压测，按峰值流量的1.5倍预留。
- CDC必须用ROW模式：否则Binlog捕获不全，数据对不上。
- Schema Registry一定要上：别图省事用JSON，后面改字段会让你崩溃。
- Filebeat内存要限制：背压机制弱，容易OOM。
- Pulsar虽好，运维成本高：小团队慎选，Kafka更成熟。
数据采集层做好了，实时OLAP系统就成功了一半。下一层，咱们聊聊数据如何高效传输与缓冲。嗯，先到这里。