4、第一个 Connector：配置 MySQL 源连接器，实现全量数据同步

好，到了真正动手的时候了。

前面我们聊了 CDC 是什么，Debezium 怎么工作的。但说实话，光看理论不过瘾。这一章，我们就来配置第一个真正的 Connector——MySQL 源连接器，把一张业务表里的数据，完整地拉到 Kafka 里来。

我个人习惯，学一个新工具，先跑通一个最简单的场景。全量同步就是那个“最简单”。它不涉及增量日志的持续监听，逻辑清晰，适合用来验证整个链路是否通畅。

4.1 准备工作：MySQL 的“体检”

在配置 Connector 之前，MySQL 那边得先做点准备。不是所有 MySQL 都能直接连 Debezium 的。我遇到过好几次，兴致勃勃配好了 Connector，结果报错说 Binlog 没开。嗯，这里要注意。

4.1.1 检查 Binlog 是否开启

Debezium 依赖 MySQL 的 Binlog。全量同步虽然不依赖 Binlog 做实时捕获，但 Debezium 的 Snapshot 机制 仍然需要读取 Binlog 位置，来保证数据一致性。所以，Binlog 必须开着。

登录 MySQL，执行这条命令：

SHOW VARIABLES LIKE 'log_bin';

如果结果是 ON，恭喜你，省了一步。如果是 OFF，那就得去改 MySQL 配置文件了。

4.1.2 配置 MySQL 用户权限

Debezium 需要一个专门的 MySQL 用户。这个用户需要哪些权限？我列个表，你直接照着建就行。

权限	作用
SELECT	读取表结构、全量数据
RELOAD	执行 FLUSH TABLES WITH READ LOCK（快照一致性）
REPLICATION CLIENT	获取 Binlog 文件名和位置
REPLICATION SLAVE	模拟 Slave 读取 Binlog
SHOW DATABASES	列举数据库（可选，但建议加上）

创建用户的 SQL 大概长这样：

CREATE USER 'debezium'@'%' IDENTIFIED BY 'your_password';
GRANT SELECT, RELOAD, REPLICATION CLIENT, REPLICATION SLAVE, SHOW DATABASES ON *.* TO 'debezium'@'%';
FLUSH PRIVILEGES;

4.1.3 准备一张测试表

我们建一张简单的订单表，用来做同步测试。

CREATE DATABASE IF NOT EXISTS demo;
USE demo;

CREATE TABLE orders (
    id INT AUTO_INCREMENT PRIMARY KEY,
    order_no VARCHAR(32) NOT NULL,
    amount DECIMAL(10,2) NOT NULL,
    status TINYINT DEFAULT 0,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

INSERT INTO orders(order_no, amount, status) VALUES
('ORD20240101001', 199.00, 1),
('ORD20240101002', 299.00, 0),
('ORD20240101003', 59.90, 1);

插三条数据进去。等会儿我们看看，这三条数据能不能完整地出现在 Kafka 里。

4.2 配置第一个 Connector：两种方式

Debezium 的 Connector 配置，说白了就是一个 JSON。你可以通过 REST API 提交，也可以写在配置文件里。我个人更推荐用 REST API，灵活，方便调试。

4.2.1 通过 REST API 提交

假设你的 Kafka Connect 运行在 localhost:8083。我们发送一个 POST 请求：

curl -i -X POST http://localhost:8083/connectors \
  -H "Content-Type: application/json" \
  -d '{
    "name": "mysql-orders-connector",
    "config": {
      "connector.class": "io.debezium.connector.mysql.MySqlConnector",
      "database.hostname": "localhost",
      "database.port": "3306",
      "database.user": "debezium",
      "database.password": "your_password",
      "database.server.id": "223344",
      "database.server.name": "dbserver1",
      "database.include.list": "demo",
      "table.include.list": "demo.orders",
      "database.history.kafka.bootstrap.servers": "localhost:9092",
      "database.history.kafka.topic": "schema-changes.demo",
      "include.schema.changes": "false",
      "snapshot.mode": "initial"
    }
  }'

这里有几个关键参数，我解释一下：

• database.server.id：Debezium 会模拟一个 MySQL Slave，这个 ID 不能和集群里其他 MySQL 节点的 ID 重复。我习惯用 223344 这种大一点的数字，不容易冲突。
• database.server.name：逻辑名称，它会作为 Kafka Topic 的前缀。比如这里配了 dbserver1，那 Topic 名就是 dbserver1.demo.orders。
• snapshot.mode：这里配 initial，意思是启动时先做一次全量快照，然后自动切换到增量模式。对于全量同步来说，这就是我们要的。

4.2.2 验证 Connector 是否启动成功

提交之后，检查一下状态：

curl http://localhost:8083/connectors/mysql-orders-connector/status

如果返回的 state 是 RUNNING，说明 Connector 已经跑起来了。如果报错，别慌，看看 trace 字段里的错误信息。我遇到过最常见的问题是权限不足，或者 Binlog 没开。

4.3 全量同步的核心：Snapshot 机制

你可能会好奇，Debezium 是怎么把 MySQL 里已有的几万条数据一次性搬出来的？

它靠的是 Snapshot（快照）。说白了，就是在某个时间点，给数据库拍一张“照片”，然后把这张照片里的数据全部发到 Kafka。

4.3.1 Snapshot 的执行流程

我画了一张图，帮你理解这个过程。

流程其实不复杂：

1. 先获取当前 Binlog 的文件名和偏移量。
2. 加一个全局读锁，确保数据一致性。
3. 读取全量数据，逐行发送到 Kafka。
4. 释放锁。

你想想看，整个过程对业务的影响其实很小。加锁的时间非常短，就是一瞬间的事。读取数据的时候，锁已经释放了。

4.3.2 全量同步的数据格式

数据到了 Kafka，长什么样？我们来消费一下看看。

kafka-console-consumer.sh \
  --bootstrap-server localhost:9092 \
  --topic dbserver1.demo.orders \
  --from-beginning \
  --property print.key=true \
  --property print.value=true

你会看到类似这样的消息：

{
    "schema": {...},
    "payload": {
        "before": null,
        "after": {
            "id": 1,
            "order_no": "ORD20240101001",
            "amount": 199.00,
            "status": 1,
            "created_at": 1704067200000
        },
        "source": {
            "version": "2.5.0.Final",
            "connector": "mysql",
            "name": "dbserver1",
            "ts_ms": 1704067200000,
            "snapshot": "true",
            "db": "demo",
            "table": "orders"
        },
        "op": "r",
        "ts_ms": 1704067200000
    }
}

注意看几个字段：

• op: "r"：表示这是一条全量快照数据（read）。增量数据会是 c（create）、u（update）、d（delete）。
• source.snapshot: "true"：明确标记这是快照数据。这个字段在后续做数据对账时非常有用。
• after：就是你的业务数据，已经转成了 JSON 格式。

4.4 避坑指南：我踩过的几个坑

配置第一个 Connector 的时候，有几个地方特别容易出问题。我把自己踩过的坑分享给你。

4.4.1 表名大小写问题

我曾经在 Linux 上部署 Debezium，MySQL 的表名是 Orders，我在配置里写的是 demo.orders（小写）。结果 Connector 一直报错说找不到表。

原因很简单：Linux 下 MySQL 的表名是大小写敏感的。而 Debezium 的 table.include.list 也是大小写敏感的。解决方案：保持大小写一致。或者，干脆统一用小写，建表时也用小写。

4.4.2 主键必须存在

Debezium 要求表必须有主键。没有主键的表，Snapshot 阶段会报错。为什么？因为 Debezium 需要靠主键来保证数据行的唯一性，尤其是在增量阶段做 UPDATE 和 DELETE 时。

如果你确实有一张没有主键的表，怎么办？我的建议是：加一个自增 ID 列做主键。这不仅是给 Debezium 用，对数据库本身也有好处。

4.4.3 时间字段的时区问题

全量同步完成后，你可能会发现 Kafka 里的时间戳和 MySQL 里的不一样。比如 MySQL 存的是 2024-01-01 10:00:00，到了 Kafka 变成了 1704067200000（UTC 时间）。

这不是 Bug。Debezium 默认会把时间字段转成 UTC 时间戳。如果你希望保持原样，可以在 Connector 配置里加上：

"converters": "datetime",
"datetime.type": "timestamp",
"datetime.format": "yyyy-MM-dd HH:mm:ss"

不过我个人建议，在消费端做时区转换更灵活。毕竟数据进了 Kafka，下游可能有多个消费者，各自的时区需求不一样。

4.5 验证全量同步结果

最后一步，验证数据对不对。

我们插入了 3 条数据，Kafka 里应该也有 3 条消息。用 kafka-console-consumer 消费一下，数一数消息条数。如果正好 3 条，每条数据的字段值都和 MySQL 里一致，那就说明全量同步成功了。

你也可以写一个简单的消费者程序，把数据打印出来，和 MySQL 里的记录逐条对比。嗯，这一步虽然简单，但能让你对整个链路建立信心。

好了，第一个 Connector 已经跑起来了。全量数据顺利进入 Kafka，你亲手搭建的 CDC 链路，通了。