📚 课程目录第3章 Debezium核心概念

3. Debezium 核心概念：Connector、Snapshot、Offset、Schema History、转换器

说实话，刚接触 Debezium 的时候，我也被这一堆术语搞得有点晕。Connector、Snapshot、Offset……每个词都认识，但串起来就不知道它们在干嘛。后来在线上环境踩了几次坑，才真正搞明白这些组件是怎么配合的。

这一章，我就带你把这几个核心概念彻底捋清楚。你理解透了它们，后面部署和运维才能心里有底。

3.1 Connector：连接器的“灵魂”

Connector 是 Debezium 里最上层的概念。说白了，它就是一个“任务定义”。你告诉 Debezium：我要监听哪个数据库、哪个表、用什么方式捕获变化。然后 Debezium 就会根据这个定义，启动一个或多个 Task 去干活。

我个人习惯把 Connector 理解成“作业单”。作业单上写清楚了数据源是谁、目标是谁、要不要做转换。Kafka Connect 集群里的 Worker 节点拿到这个作业单，就开始执行。

  关键点： Connector 本身不干活，它只负责描述“干什么活”。真正干活的是它创建的 Task。

举个例子，一个 MySQL Connector 的配置大概长这样：

{
  "name": "mysql-connector-orders",
  "config": {
    "connector.class": "io.debezium.connector.mysql.MySqlConnector",
    "database.hostname": "192.168.1.100",
    "database.port": "3306",
    "database.user": "debezium",
    "database.password": "debezium_pwd",
    "database.server.id": "184054",
    "database.server.name": "dbserver1",
    "database.include.list": "order_db",
    "table.include.list": "order_db.orders, order_db.order_items",
    "database.history.kafka.bootstrap.servers": "kafka1:9092,kafka2:9092",
    "database.history.kafka.topic": "schema-changes.order_db",
    "include.schema.changes": "true"
  }
}

你看，这里指定了连接哪个库、监听哪些表、Schema 历史存到哪个 Kafka Topic。这些信息一旦提交，Kafka Connect 就会创建对应的 Task 开始拉取 binlog。

小技巧： 我建议你在生产环境里，每个 Connector 只监听一个业务库。别图省事把所有表塞到一个 Connector 里，出了问题排查起来很痛苦。

3.2 Snapshot：初次见面的“全量快照”

当一个 Connector 第一次启动时，它面对的是一个“陌生”的数据库。它不知道表里原来有什么数据。这时候就需要 Snapshot 机制出场了。

Snapshot 就是 Debezium 在启动时，对监听的表做一次全量读取。它会先把表里已有的所有数据读出来，一条一条发送到 Kafka。等全量数据发完了，再切换到增量模式，监听实时的 binlog 变更。

为什么会这样设计？你想想看，如果你只监听增量，那之前的数据就永远拿不到了。对于初次建立数据管道来说，全量+增量才是完整的方案。

Snapshot 有几种模式，我整理了一个表格：

模式	行为	适用场景
initial	先执行全量快照，再切换到增量	默认模式，初次部署时使用
when_needed	只在需要时执行快照（比如 Schema 变更后）	已有数据管道，仅需补充
never	跳过全量快照，直接开始增量	只关心增量数据，不关心历史
initial_only	只做全量快照，做完就停止	一次性数据迁移

注意： 我曾经在生产环境遇到过一个问题——大表做 Snapshot 时，因为数据量太大（几亿行），直接把 Kafka 集群打满了。后来我加了 snapshot.fetch.size 参数，把每次拉取的行数调小，才稳住。大表快照一定要做限流。

3.3 Offset：断点续传的“书签”

Offset 是 Debezium 里最容易被忽视、但最重要的概念。没有它，一旦进程重启，你就得从头开始读 binlog——那数据就全乱套了。

Offset 记录了 Debezium 当前已经读到 binlog 的哪个位置。它通常包含 binlog 文件名、偏移量、事件时间戳等信息。每次处理完一条变更事件，Debezium 就会把最新的 Offset 提交到 Kafka 的一个内部 Topic（通常是 connect-offsets）。

当 Connector 重启时，它会先去查 Offset，找到上次中断的位置，然后从那里继续读。这就是“断点续传”的核心机制。

我举个例子，一个 MySQL 的 Offset 记录大概长这样：

{
  "file": "mysql-bin.000123",
  "pos": 456789,
  "gtids": "d4b8e8f0-1a2b-3c4d-5e6f-7a8b9c0d1e2f:1-12345",
  "event_ts_usec": 1698765432000000
}

  核心原则： Offset 是 Debezium 的“命根子”。如果 Offset 丢失或损坏，Connector 只能重新做 Snapshot，或者从 binlog 的开头重新读。所以，千万不要手动删除 Offset Topic。

我记得有一次，运维同事不小心清空了 connect-offsets Topic，结果所有 Connector 都重新开始读 binlog，导致下游 Kafka Topic 里出现了大量重复数据。那次事故让我养成了一个习惯——每次操作 Offset 相关配置前，先备份。

3.4 Schema History：Schema 演变的“档案室”

数据库的 Schema 不是一成不变的。今天加个字段，明天改个类型，后天删个列。Debezium 必须知道这些变化，否则它解析出来的数据可能对不上号。

Schema History 就是用来记录这些 Schema 变更的。它会把所有 DDL 语句（比如 ALTER TABLE）都存到一个专门的 Kafka Topic 里。当 Debezium 启动时，它会先从这个 Topic 里回放所有的 Schema 变更，重建出当前的表结构，然后再开始解析 binlog。

没有 Schema History 会怎样？我告诉你，后果很严重。如果 Debezium 不知道表结构已经变了，它可能把新字段的数据解析成乱码，甚至直接报错崩溃。

配置 Schema History 很简单，就是在 Connector 配置里指定一个 Kafka Topic：

"database.history.kafka.topic": "schema-changes.order_db"

建议： 每个 Connector 使用独立的 Schema History Topic，命名规则建议用 schema-changes.<database_name>。这样方便管理和排查。

3.5 转换器：数据格式的“翻译官”

Debezium 从 binlog 里读到的数据是数据库原生的格式。但下游系统不一定能直接消费这种格式。这时候就需要转换器（Converter）出场了。

转换器负责把 Debezium 的内部数据结构，转换成下游系统能识别的格式。最常见的两种转换器是：

JSON Converter：把数据转成 JSON 格式，可读性好，适合调试和简单场景
Avro Converter：把数据转成 Avro 格式，体积小、性能高、带 Schema 管理，适合生产环境

我个人强烈建议生产环境用 Avro。为什么？因为 JSON 不带 Schema 信息，下游消费者如果不知道字段类型，很容易解析出错。而 Avro 自带 Schema Registry，每个字段的类型、默认值都清清楚楚。

配置转换器也很简单，在 Kafka Connect 的 worker 配置文件里设置：

key.converter=org.apache.kafka.connect.json.JsonConverter
value.converter=org.apache.kafka.connect.json.JsonConverter
key.converter.schemas.enable=true
value.converter.schemas.enable=true

或者用 Avro：

key.converter=io.confluent.connect.avro.AvroConverter
value.converter=io.confluent.connect.avro.AvroConverter
key.converter.schema.registry.url=http://schema-registry:8081
value.converter.schema.registry.url=http://schema-registry:8081

避坑指南： 我曾经遇到过一个坑——JSON 模式下 schemas.enable 设为 false，结果下游消费者拿到的数据没有 Schema 信息，字段类型全靠猜。后来排查了半天才发现是这里的问题。所以，要么用 Avro，要么把 JSON 的 Schema 开关打开。

3.6 核心概念关系图

说了这么多，你可能有点乱。没关系，我画了一张图，把这些概念的关系串起来：

从这张图你可以看到，整个流程是：源数据库产生变更事件 → Connector 接收并解析 → 转换器格式化 → 最终写入 Kafka Topic。而 Snapshot、Offset、Schema History 这三个组件，就像三个“后勤部门”，在背后默默支撑着整个流程的稳定运行。

3.7 小结

好了，这一章的内容就这么多。我们来快速回顾一下：

Connector 是作业定义，描述监听谁、怎么监听
Snapshot 负责初次的全量数据读取
Offset 记录断点位置，保证重启后不丢数据
Schema History 记录表结构变更，保证解析不出错
转换器 把数据转成下游能识别的格式