4. 流式数据采集：Kafka Connect原理、Debezium CDC实战、实时元数据变更捕获

好，我们进入第四章。流式数据采集，说白了就是让数据自己流进来，而不是你半夜爬起来跑批处理脚本。我个人觉得，这是构建实时数据目录最关键的一环——你连数据源的变化都抓不住，那目录就是个死目录。

这一章我们聚焦三个东西：Kafka Connect 到底怎么工作的、Debezium 做 CDC 的实战套路、以及怎么把元数据的变更实时捕获到目录里。嗯，开始吧。

4.1 Kafka Connect 原理：连接器、转换器与 Worker

Kafka Connect 是 Kafka 生态里的一个组件，专门用来跟外部系统打交道。你想想看，如果没有它，你要自己写 Producer 去读数据库、写 Consumer 去推数据到 Elasticsearch，那得写多少重复代码？

Kafka Connect 把这件事抽象成了三个核心概念：

• Connector（连接器）：负责跟外部系统交互。比如 Source Connector 从数据库拉数据，Sink Connector 把数据推到 HDFS。
• Converter（转换器）：负责数据格式的序列化/反序列化。常见的有 JSON、Avro、Protobuf。我建议你用 Avro，带 Schema 注册表，后面做元数据管理会省很多事。
• Worker（工作节点）：真正干活的那个进程。可以单机跑，也可以集群部署。集群模式下，Connector 和 Task 会被自动分配到不同的 Worker 上。

这里有个关键点：Task 是并行度的体现。一个 Connector 可以拆成多个 Task，每个 Task 处理一部分数据分片。比如你读一张 1 亿行的表，拆成 4 个 Task，每个读 2500 万行，速度就上去了。

我在项目中遇到过一个问题：某个团队把 Converter 配成了 JSON，结果字段类型全变成了字符串，下游做聚合时疯狂报错。后来改成 Avro + Schema Registry，问题才解决。所以，Converter 的选择直接影响数据质量。

4.2 Debezium CDC 实战：从 MySQL 到 Kafka

CDC（Change Data Capture）是流式数据采集的核心技术。Debezium 是目前最主流的 CDC 工具之一，基于 Kafka Connect 架构，专门用来捕获数据库的变更事件。

它的原理其实不复杂：Debezium 模拟成一个 MySQL Slave，读取 Binlog 的变更事件，然后转换成 Kafka 消息。每条消息包含三个部分：before（变更前）、after（变更后）、op（操作类型）。

操作类型有四种：

操作类型	含义	典型场景
c	Create（创建）	新增一条记录
u	Update（更新）	修改字段值
d	Delete（删除）	删除一条记录
r	Snapshot（快照）	初始全量读取

下面是一个典型的 Debezium 配置示例，用来监听 MySQL 的 inventory 数据库：

{
  "name": "inventory-connector",
  "config": {
    "connector.class": "io.debezium.connector.mysql.MySqlConnector",
    "database.hostname": "192.168.1.100",
    "database.port": "3306",
    "database.user": "debezium",
    "database.password": "dbz_pass",
    "database.server.id": "184054",
    "database.server.name": "dbserver1",
    "database.include.list": "inventory",
    "database.history.kafka.bootstrap.servers": "kafka:9092",
    "database.history.kafka.topic": "schema-changes.inventory",
    "include.schema.changes": "true",
    "key.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter.schemas.enable": "false"
  }
}

配置好之后，Debezium 会自动做两件事：

1. 先做一次快照：把当前表的所有数据读出来，每条记录生成一个 op: "r" 的消息。
2. 然后持续监听：后续的增删改都会实时生成变更事件。

4.3 实时元数据变更捕获：从 Binlog 到数据目录

好，现在数据变更已经能实时流到 Kafka 了。但我们的目标是构建数据目录，所以还需要把元数据的变更也捕获进来。

什么是元数据变更？比如：

• 新增了一张表
• 某个字段改了类型
• 删除了一个索引
• 修改了表注释

这些变更不会出现在业务数据的 Binlog 里，它们出现在 DDL 语句 中。Debezium 提供了一个 database.history.kafka.topic 参数，专门用来记录 Schema 变更历史。

我建议你这样做：

1. 开启 Schema 变更捕获：在 Debezium 配置里设置 include.schema.changes: true。
2. 消费 Schema 变更 Topic：写一个消费者，监听 schema-changes.inventory 这个 Topic。
3. 解析 DDL 事件：每条消息里包含 ddl 字段，就是原始的 SQL 语句。
4. 更新数据目录：把解析出来的表名、字段名、类型、注释等信息，写入数据目录的元数据存储。

下面是一个简化的 DDL 事件消息示例：

{
  "source": {
    "server": "dbserver1",
    "db": "inventory",
    "table": "orders"
  },
  "position": {
    "ts_sec": 1700000000,
    "file": "mysql-bin.000123",
    "pos": 45678
  },
  "ddl": "ALTER TABLE orders MODIFY COLUMN status VARCHAR(20) NOT NULL DEFAULT 'pending'",
  "tableChanges": [
    {
      "type": "ALTER",
      "id": "\"inventory\".\"orders\"",
      "table": {
        "defaultCharsetName": "utf8mb4",
        "columns": [
          {
            "name": "status",
            "jdbcType": 12,
            "typeName": "VARCHAR",
            "length": 20,
            "nullable": false,
            "position": 5,
            "defaultValueExpression": "'pending'"
          }
        ]
      }
    }
  ]
}

看到 tableChanges 字段了吗？它已经把 DDL 解析成了结构化的列信息。你直接拿这个去更新数据目录就行，不用自己写 SQL 解析器。

4.4 整体架构图

下面这张图展示了从 MySQL Binlog 到数据目录的完整链路。我把它画出来，方便你理解各个组件之间的关系。

从图上你可以看到，整个链路分两条线：

• 实线（主数据流）：MySQL → Debezium → Kafka → 数据流消费者 → 数据目录（血缘）。
• 虚线（元数据流）：Kafka Schema Topic → 元数据消费者 → 数据目录（元数据更新）。

这两条线是并行的，互不阻塞。但要注意，元数据更新必须比业务数据先到达目录，否则下游查询会拿到过时的 Schema 信息。

4.5 避坑指南

最后，分享几个我在实战中踩过的坑：

• Binlog 格式必须为 ROW：Debezium 只支持 ROW 格式的 Binlog。如果数据库配的是 STATEMENT 或 MIXED，CDC 会直接挂掉。我曾经接手过一个项目，Binlog 配错了，Debezium 跑了三天没报错，但一条变更都没抓到。
• 注意 Kafka 消息大小限制：如果一张表有 BLOB 或 TEXT 字段，单条消息可能超过默认的 1MB 限制。记得调大 max.message.bytes 和 replica.fetch.max.bytes。
• Schema 变更 Topic 不要随便删：Debezium 启动时需要从 Schema Topic 恢复历史 DDL。如果你删了它，重启后 Debezium 会认为所有表都是新的，重新做一次全量快照。嗯，那场面相当壮观。
• 监控延迟：用 Kafka Lag 监控工具（比如 Burrow）盯着 CDC 的消费延迟。一旦延迟超过 5 分钟，说明 Binlog 堆积了，得赶紧扩容 Task 或优化消费逻辑。