4. 增量数据源接入：四种主流方式实战
增量计算的第一步，就是搞定数据源。数据进不来，后面再牛的计算引擎也是白搭。
我这些年接过的增量数据源，总结下来就四种：MySQL Binlog、Kafka 消息队列、日志文件、API 拉取。每种都有各自的脾气，咱们一个一个说。
4.1 MySQL Binlog 解析与接入
Binlog 是 MySQL 的二进制日志，记录了所有数据变更。说白了，它就是数据库的「黑匣子」。我最早接触增量计算时，用的就是 Binlog。
4.1.1 工作原理
MySQL 主库在执行 INSERT、UPDATE、DELETE 时，会顺带写 Binlog。我们通过模拟 Slave 节点，就能实时拿到这些变更事件。
核心流程：
- 应用伪装成 MySQL Slave，向主库发送 COM_REGISTER_SLAVE 请求
- 主库开始推送 Binlog 事件流
- 应用解析事件，提取行变更数据
- 将解析后的数据写入 Kafka 或直接入仓
4.1.2 常用工具对比
| 工具 | 语言 | 优点 | 缺点 |
|---|---|---|---|
| Canal | Java | 成熟稳定，社区活跃 | 依赖 Java 环境 |
| Maxwell | Java | 输出 JSON，轻量 | 不支持全量初始化 |
| Debezium | Java | Kafka Connect 原生集成 | 配置较复杂 |
| go-mysql-elasticsearch | Go | 性能好，资源占用低 | 功能相对单一 |
我个人习惯用 Canal。为什么？因为我在项目中踩过 Maxwell 的坑——有一次线上 Binlog 文件轮换时，Maxwell 丢了几条数据，排查了半天才发现是 checkpoint 机制的问题。Canal 在这方面更稳。
4.1.3 接入示例（Canal）
// Canal 客户端接入示例
CanalConnector connector = CanalConnectors.newSingleConnector(
new InetSocketAddress("127.0.0.1", 11111),
"example", "", ""
);
connector.connect();
connector.subscribe(".*\\..*");
connector.rollback();
while (running) {
Message message = connector.getWithoutAck(100);
long batchId = message.getId();
if (batchId == -1 || message.getEntries().isEmpty()) {
Thread.sleep(100);
continue;
}
for (CanalEntry.Entry entry : message.getEntries()) {
if (entry.getEntryType() == CanalEntry.EntryType.ROWDATA) {
// 解析行数据变更
CanalEntry.RowChange rowChange = CanalEntry.RowChange.parseFrom(
entry.getStoreValue()
);
// 处理 INSERT / UPDATE / DELETE 事件
}
}
connector.ack(batchId);
}
注意：Binlog 有三种格式：STATEMENT、ROW、MIXED。增量计算必须用 ROW 格式，否则拿不到变更前后的具体字段值。我曾经接手过一个项目，用的 STATEMENT 格式，结果解析出来的全是 SQL 语句，根本没法用。
4.2 Kafka 消息队列接入
Kafka 是增量计算的中枢神经。数据从各个源进来，经过 Kafka 缓冲，再被下游消费。你想想看，如果没有 Kafka，所有消费者都直接怼数据库，那数据库早就挂了。
4.2.1 为什么选 Kafka
- 高吞吐：单机每秒几十万条消息，轻松应对业务高峰
- 持久化：消息落盘，消费者挂了也不丢数据
- 回溯消费：可以重置 offset，重新消费历史数据
- 解耦：生产者和消费者互不依赖
4.2.2 接入策略
我建议用 Kafka Connect 作为接入框架。它自带 Source 和 Sink 概念，配置起来很顺手。
# Kafka Connect Source 配置示例
name=mysql-binlog-source
connector.class=io.debezium.connector.mysql.MySqlConnector
database.hostname=192.168.1.100
database.port=3306
database.user=canal
database.password=canal123
database.server.name=my-app-connector
database.include.list=mydb
table.include.list=mydb.orders,mydb.users
offset.storage.file.filename=/tmp/offsets.dat
小技巧：Kafka 的 topic 分区数要提前规划好。我一般按表的主键哈希来分区，保证同一行数据的变更事件落到同一个分区，这样下游消费时能保证顺序。
4.2.3 消费端注意事项
消费 Kafka 时，最容易出问题的是 offset 管理。我曾经遇到过：消费者处理完数据但还没来得及提交 offset，进程就挂了。重启后重复消费，导致下游数据重复。
解决办法是：先提交 offset，再处理业务逻辑。或者用 Kafka 的事务 API 保证 exactly-once 语义。
4.3 日志文件增量采集
很多老系统没有 Binlog，也没有消息队列，只有日志文件。这时候就得靠日志采集工具了。
4.3.1 采集原理
日志文件增量采集的核心是「文件指针」。工具会记住上次读到了哪个位置，下次从那里继续读。就像你看书夹书签一样。
4.3.2 常用工具
| 工具 | 特点 | 适用场景 |
|---|---|---|
| Filebeat | 轻量，资源占用低 | 单机日志采集 |
| Flume | 功能丰富，支持多种 Sink | 复杂日志管道 |
| Logstash | 强大的过滤和解析能力 | 需要数据清洗的场景 |
| Fluentd | 插件生态丰富 | 容器化环境 |
4.3.3 避坑指南
我曾经在日志采集上栽过跟头。那次是日志文件按天滚动，Filebeat 的配置没处理好，导致第二天采集了新文件，但旧文件的增量数据全丢了。
后来我总结了几条经验：
- 一定要配置好文件轮换策略，确保旧文件被完整采集
- 日志格式要统一，最好用 JSON 格式，解析起来方便
- 采集进度要持久化，防止进程重启后从头开始
4.4 API 增量拉取策略
最后一种，也是最头疼的一种——API 拉取。为什么头疼？因为 API 不是咱们自己控制的，人家给什么数据，咱们就得用什么数据。
4.4.1 常见拉取模式
- 基于时间戳：每次请求带上上次拉取的时间，服务端返回该时间之后的增量数据
- 基于游标：服务端返回一个游标（cursor），下次请求带上这个游标继续拉
- 基于偏移量：类似分页，每次指定偏移量和条数
- 基于事件 ID：服务端为每条数据生成递增 ID，客户端记录已拉取的最大 ID
4.4.2 策略选择
| 策略 | 优点 | 缺点 |
|---|---|---|
| 时间戳 | 实现简单 | 时间精度问题，可能漏数据 |
| 游标 | 稳定可靠，不会漏数据 | 服务端需要维护游标状态 |
| 偏移量 | 适合静态数据集 | 数据频繁增删时不准 |
| 事件 ID | 天然支持增量 | 依赖服务端生成有序 ID |
我个人最推荐游标模式。为什么？因为时间戳模式有个致命问题：如果同一秒内有多条数据，你很难保证全部拉到。游标模式就不存在这个问题，它保证每条数据只被拉取一次。
4.4.3 实现示例
# Python 游标拉取示例
def pull_incremental_data(api_url, cursor=None):
params = {"limit": 1000}
if cursor:
params["cursor"] = cursor
response = requests.get(api_url, params=params)
data = response.json()
# 处理数据
process_data(data["items"])
# 判断是否还有下一页
if data.get("next_cursor"):
pull_incremental_data(api_url, data["next_cursor"])
注意：API 拉取一定要加重试机制和退避策略。我遇到过 API 限流，连续请求 10 次后直接返回 429。后来加了指数退避，每次失败后等待时间翻倍，问题就解决了。
4.5 四种接入方式对比总结
说了这么多，咱们来张图总结一下四种方式的适用场景。
嗯，四种方式各有千秋。实际项目中，我经常是混合使用的。比如核心业务表用 Binlog，外部合作数据用 API，老系统日志用 Filebeat。所有数据最终汇入 Kafka，再由 OLAP 引擎消费。
记住一点：增量接入不是技术选型比赛，而是解决实际问题。哪种方式能让数据稳定、及时地进入数仓，就用哪种。