📚 课程目录第2章实时数据采集技术选型

实时数据采集技术选型：Canal原理与部署、Debezium配置、Kafka Connect使用

数据采集是实时数仓的入口。入口没选好，后面全是坑。

我见过不少团队，ODS层还没建起来，就先在采集上折腾了两周。说白了，选型不是选最火的，而是选最匹配你业务场景的。今天我们就聊三个主流工具：Canal、Debezium、Kafka Connect。

核心观点：实时采集的本质是“监听变更”。不管是MySQL的binlog，还是PostgreSQL的WAL，我们要做的就是把数据库的每一次INSERT、UPDATE、DELETE抓出来，转成消息，扔进Kafka。

Canal：阿里出品的binlog搬运工

Canal是阿里巴巴开源的项目。它的核心逻辑很简单：伪装成MySQL的从库，拉取binlog，然后解析成你能用的格式。

原理一句话概括：Canal把自己注册为MySQL Slave，主库把binlog推给它，它解析后推给Kafka或RocketMQ。

我的经验：我第一次用Canal时，以为部署完就能跑。结果发现MySQL必须开启binlog，而且格式得是ROW。嗯，这个坑我踩过，后来写了个checklist，每次部署前先跑一遍。

Canal部署步骤

MySQL侧准备：开启binlog，设置binlog_format=ROW，创建Canal专用账号并授予REPLICATION SLAVE和REPLICATION CLIENT权限。
下载解压：从GitHub Release页下载Canal Deployer，解压到服务器。
配置canal.properties：设置Canal Server模式（tcp/kafka/rocketmq），指定Kafka地址。
配置instance.properties：指定要监听的数据库、表名（支持正则），设置目标Topic。
启动验证：运行bin/startup.sh，查看logs/example/example.log确认连接成功。

# canal.properties 关键配置
canal.serverMode = kafka
canal.mq.servers = 192.168.1.10:9092,192.168.1.11:9092

# instance.properties 关键配置
canal.instance.master.address = 192.168.1.20:3306
canal.instance.dbUsername = canal_user
canal.instance.dbPassword = canal_pass
canal.instance.filter.regex = mydb\\.user_table
canal.mq.topic = ods_user_binlog

注意：Canal默认只解析DML（INSERT/UPDATE/DELETE）。如果你需要DDL变更记录，得额外配置canal.instance.filter.query.dml.threshold。我曾经因为没配这个，丢了一批表结构变更事件，导致下游Schema对不上。

Debezium：CDC领域的瑞士军刀

Debezium是Red Hat开源的项目。它基于Kafka Connect架构，支持MySQL、PostgreSQL、MongoDB、SQL Server等多种数据库。

我个人更喜欢Debezium的一点：它把每条变更记录包装成结构化的消息，包含before、after、source元数据。你想想看，这对实时数仓的ODS层来说，简直是量身定做。

Debezium配置示例（MySQL Connector）

{
  "name": "mysql-connector-01",
  "config": {
    "connector.class": "io.debezium.connector.mysql.MySqlConnector",
    "database.hostname": "192.168.1.20",
    "database.port": "3306",
    "database.user": "debezium_user",
    "database.password": "debezium_pass",
    "database.server.id": "223344",
    "database.server.name": "myapp",
    "database.include.list": "mydb",
    "table.include.list": "mydb.user,mydb.order",
    "database.history.kafka.bootstrap.servers": "192.168.1.10:9092",
    "database.history.kafka.topic": "schema-changes.mydb",
    "include.schema.changes": "true",
    "transforms": "unwrap",
    "transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState"
  }
}

配置里有个关键点：database.server.name。这个值会作为Kafka Topic的前缀。比如你设成myapp，那Topic名就是myapp.mydb.user。我建议你提前规划好命名规范，不然后面Topic一多，找起来很痛苦。

特性	Canal	Debezium
数据库支持	仅MySQL	MySQL、PG、MongoDB、SQL Server等
消息格式	扁平JSON（可自定义）	结构化Avro/JSON（含before/after/source）
Schema演进	需手动处理	内置Schema Registry支持
部署复杂度	独立服务，配置简单	依赖Kafka Connect集群
社区活跃度	国内活跃	全球活跃，Confluent官方支持

Kafka Connect：插件化的数据管道

Kafka Connect是Apache Kafka自带的组件。它不是一个采集工具，而是一个框架。你只需要写（或下载）一个Connector插件，就能把数据从任意源拉到Kafka。

我习惯把Kafka Connect看作“乐高底座”。Debezium就是搭在上面的一个插件。你还可以装JDBC Connector、Elasticsearch Connector、S3 Connector……说白了，只要社区有插件，你就不用自己写代码。

Kafka Connect的两种模式

Standalone模式：单进程运行，适合开发和测试。配置都在一个文件里，启动简单。
Distributed模式：多节点集群，支持自动容错和负载均衡。生产环境必须用这个。

我的建议：别在Standalone模式上花太多时间。我见过有人用Standalone跑了一个月，某天进程挂了，数据全丢了。直接上Distributed模式，配合REST API管理Connector，这才是正经做法。

# 启动Distributed模式
bin/connect-distributed.sh config/connect-distributed.properties

# 通过REST API创建Connector
curl -X POST -H "Content-Type: application/json" \
  --data @mysql-connector.json \
  http://localhost:8083/connectors

# 查看Connector状态
curl http://localhost:8083/connectors/mysql-connector-01/status

选型决策：到底用哪个？

这个问题没有标准答案。我根据项目经验，给你几个判断维度：

如果你只用MySQL，团队对Java比较熟：选Canal。部署轻量，国内文档多，遇到问题好找人问。
如果你有多数据库，或者需要Avro/Protobuf序列化：选Debezium。它的消息结构更规范，配合Schema Registry能自动处理字段变更。
如果你已经上了Kafka生态，不想再维护一套独立服务：选Kafka Connect + Debezium插件。统一管理，运维成本低。

一句话总结：Canal适合“快速搞定MySQL”，Debezium适合“规范化CDC”，Kafka Connect适合“统一数据管道”。三者不是互斥的，我见过不少团队用Kafka Connect跑Debezium，同时用Canal做辅助通道。

嗯，采集层选型就聊到这儿。工具只是手段，关键是理解binlog的流转逻辑。下一层我们就要开始处理这些原始数据了——DWD层的清洗和转换，那才是真正出活儿的地方。