📚 课程目录第4章数据源接入与采集

4. 数据源接入与采集：从源头说起

数据采集这件事，说白了就是「怎么把数据从别人家搬到你家」。我做了这么多年实时系统，见过太多项目在采集阶段就翻车了。不是选型不对，就是格式没想清楚，最后数据到了，解析不了，白忙活。

这一章，咱们就把常见数据源、采集方案、数据格式这三个问题聊透。你想想看，源头没搞定，后面再好的架构也是空中楼阁。

4.1 常见数据源：你大概率会遇到这四种

实时数据源，我归纳下来就四大类。每一类的脾气都不一样，得顺着毛捋。

4.1.1 MySQL Binlog：数据库变更的「流水账」

MySQL 的 Binlog（二进制日志），说白了就是数据库的「黑匣子」。每次 INSERT、UPDATE、DELETE 都会记在里面。实时同步 MySQL 数据到其他系统，基本都靠它。

我记得有一次，业务方要求订单数据变更后 3 秒内同步到 Elasticsearch。我们就是靠解析 Binlog 做到的。Binlog 有三种格式：

格式	说明	我建议
STATEMENT	记录 SQL 语句	别用，主从复制容易不一致
ROW	记录每行变更前后数据	实时采集首选，信息完整
MIXED	混合模式	看场景，但 ROW 更省心

我的习惯：生产环境 Binlog 格式一律用 ROW。虽然日志量会大一些，但解析出来的数据最干净，不会出现「明明改了 10 条，只同步了 5 条」的尴尬。

4.1.2 日志：最古老也最通用的数据源

日志文件，每个系统都有。Nginx 访问日志、Java 应用日志、系统日志…… 实时采集日志，核心就是「盯着文件尾部，有新内容就立刻读走」。

嗯，这里要注意：日志采集最怕日志轮转（log rotation）。你正读着文件呢，系统把文件重命名了，新日志写到新文件里去了。采集器如果没处理好，就会丢数据。

4.1.3 API：对接外部系统的「标准接口」

很多第三方服务只提供 HTTP API 给你拉数据。比如支付回调、社交媒体数据、天气数据等等。API 采集分两种：

推模式（Webhook）： 对方有数据变更就主动调你的接口。实时性最好，但你需要暴露公网接口。
拉模式（Polling）： 你定时去调对方的接口。实现简单，但实时性取决于轮询间隔。

我在项目中遇到过一个问题：拉模式轮询间隔设得太短，把对方 API 打挂了。后来我们加了退避策略，每次请求失败后自动延长间隔。

4.1.4 IoT：海量设备的「数据洪流」

IoT 设备的数据，特点是量大、频率高、格式乱。温度传感器、GPS 定位、设备状态…… 每秒可能涌入几十万条消息。这类数据源，通常走 MQTT 或 CoAP 协议，消息体很小，但数量巨大。

避坑指南：我曾经接手过一个 IoT 项目，设备上报的数据里混着乱码和空值。后来我们强制要求设备端做数据校验，采集层也加了过滤规则。记住：IoT 设备不可信，数据必须清洗。

4.2 采集方案选型：工具选对了，事半功倍

数据源搞清楚了，接下来就是选采集工具。我见过不少团队，上来就自己写采集脚本，结果维护成本高得吓人。其实社区里已经有很成熟的方案了。

4.2.1 Canal：MySQL Binlog 采集的「标配」

Canal 是阿里巴巴开源的项目，专门用来解析 MySQL Binlog。它的工作原理很简单：伪装成 MySQL 的从库，拉取 Binlog 然后解析成你需要的格式。

我个人习惯用 Canal 1.1.5 以上版本，支持直接投递到 Kafka 或 RocketMQ。部署也很轻量，一个 Java 进程搞定。

# Canal 配置示例：监听某个数据库的所有表
canal.instance.master.address=192.168.1.100:3306
canal.instance.dbUsername=canal
canal.instance.dbPassword=canal@123
canal.instance.filter.regex=order_db\\..*
canal.mq.topic=order_binlog_topic

  核心要点：Canal 只负责「拉」和「解析」，不负责「存」。解析后的数据建议直接进消息队列，不要落盘。这样下游可以灵活消费。

4.2.2 Flume：日志采集的「瑞士军刀」

Flume 是 Apache 的项目，专门用来采集、聚合、传输海量日志。它的架构是 Source → Channel → Sink，每一层都可以自定义。

你想想看，几十台服务器的日志要统一采集到 HDFS 或 Kafka，用 Flume 最合适。它支持 Taildir Source，可以监控多个文件目录，还能记录读取位置，重启后不丢数据。

组件	作用	常用实现
Source	从哪里读数据	Taildir Source、Kafka Source
Channel	临时存储数据	Memory Channel、File Channel
Sink	数据写到哪	Kafka Sink、HDFS Sink

我的建议：生产环境 Channel 用 File Channel，虽然慢一点，但不会丢数据。Memory Channel 快，但进程挂了数据就没了。看你要速度还是要可靠性。

4.2.3 SDK：自己动手，丰衣足食

有些场景，Canal 和 Flume 都搞不定。比如你要采集某个私有协议的数据，或者数据源是自研的中间件。这时候就得自己写采集 SDK。

写 SDK 有几个原则：

轻量： 不要引入太多依赖，一个 HTTP 客户端 + 序列化库就够了。
断线重连： 网络抖动是常态，SDK 必须能自动重连。
背压机制： 下游处理不过来时，SDK 要能主动降速，别把对方打爆。

我记得有一次，我们自研的 SDK 没有做背压，结果数据量一上来，直接把采集服务的内存撑爆了。后来加了基于令牌桶的限流，才稳住。

4.3 数据格式与协议：统一语言，避免鸡同鸭讲

数据采集上来，用什么格式传输？这个问题不解决，后面解析的时候会疯掉。我见过一个团队，同一个数据源，JSON、XML、CSV 混着用，下游消费方苦不堪言。

4.3.1 JSON：简单直观，但别滥用

JSON 是大家最熟悉的格式。可读性好，调试方便。但它的缺点也很明显：

体积大： 同样的数据，JSON 比二进制格式大 3-5 倍。
解析慢： 字符串解析比二进制解析慢一个数量级。
无 schema： 字段类型全靠约定，容易出错。

我的习惯是：调试和开发环境用 JSON，生产环境尽量别用。尤其是高吞吐场景，JSON 的带宽和 CPU 开销会让你肉疼。

4.3.2 Avro：大数据生态的「通用语言」

Avro 是 Hadoop 生态里的序列化框架。它用 JSON 定义 schema，数据存成二进制。好处是：

体积小： 比 JSON 小 60% 以上。
带 schema： 解析时自动校验字段类型。
支持 schema 演化： 新增字段不影响旧数据。

{
  "type": "record",
  "name": "Order",
  "fields": [
    {"name": "order_id", "type": "string"},
    {"name": "amount", "type": "double"},
    {"name": "status", "type": "int"}
  ]
}

  核心要点：Avro 的 schema 要统一管理，建议放在 Schema Registry 里。这样生产者和消费者都从 Registry 拉 schema，保证两边一致。

4.3.3 Protobuf：高性能场景的「首选」

Protobuf 是 Google 的序列化协议。它的特点是：

极致性能： 序列化/反序列化速度比 JSON 快 10 倍以上。
强类型： .proto 文件定义了严格的字段类型和编号。
向后兼容： 新增字段不影响旧版本。

嗯，这里要注意：Protobuf 的 .proto 文件需要编译成对应语言的代码。如果团队里语言多，维护 .proto 文件会有点麻烦。

我曾经在一个 IoT 项目里用 Protobuf，设备端用 C++，服务端用 Java。.proto 文件统一维护在 Git 仓库里，每次修改后自动编译发布。效果很好，数据量从每秒 10 万条降到 2 万条（因为体积小了）。

4.4 知识体系总览

说了这么多，我画了一张图帮你理清思路。数据源、采集方案、数据格式这三层的关系，一目了然。

这张图把数据源、采集方案、数据格式串起来了。你从左边选数据源，中间选采集工具，右边定数据格式。一条链路走到底，别混搭。

  总结一下：数据采集没有银弹。MySQL 同步用 Canal，日志采集用 Flume，特殊场景自己写 SDK。格式上，调试用 JSON，大数据用 Avro，高性能用 Protobuf。记住一个原则：源头统一，格式统一，后面才省心。