使用Debezium进行MySQL变更数据捕获（CDC）实战

Debezium通过监听MySQL binlog实现数据实时同步，需配置MySQL、部署Connector、设置Kafka Connect并消费变更事件；选择合适配置需根据需求设定server.id、连接信息、包含/排除表及快照模式；变更事件以JSON格式发布至Kafka，含before、after、op等字段，下游应用解析后执行对应操作；可通过Kafka Streams或Flink处理；使用Kafka Connect REST API和JMX指标监控Connector状态与性能；Schema演化通过Schema History Topic和注册表（如Confluent Schema Registry）管理；初始快照可配置模式与锁策略以减少数据库压力；性能优化包括提升资源、调整参数与数据库配置；数据一致性可通过事务性Outbox、Heartbeat、Kafka事务及数据比对保障。

Debezium通过捕获MySQL的变更数据，可以实时同步数据到其他系统，实现数据集成和微服务架构。它监听MySQL的binlog，将数据变更转化为事件流，供下游应用消费。

使用Debezium进行MySQL CDC实战主要涉及配置MySQL、部署Debezium Connector、配置Kafka Connect以及消费变更事件。

配置MySQL以启用binlog。 部署Debezium Connector到Kafka Connect集群。 配置Connector以连接到MySQL数据库并指定要捕获的数据库和表。 下游应用通过Kafka消费变更事件。

如何选择合适的Debezium Connector配置？

选择合适的Debezium Connector配置取决于你的具体需求。关键配置包括：

• database.server.id

  : MySQL服务器的唯一ID，确保在集群中唯一。

• database.hostname

  和

  database.port

  : MySQL服务器的地址和端口。

• database.user

  和

  database.password

  : 用于连接MySQL的用户名和密码，需要具有足够的权限读取binlog。

• database.include.list

  和

  database.exclude.list

  : 指定要捕获或排除的数据库列表。

• table.include.list

  和

  table.exclude.list

  : 指定要捕获或排除的表列表。

• snapshot.mode

  : 定义初始快照模式，例如

  initial

  （首次启动时执行快照）或

  never

  （不执行快照）。

• topic.prefix

  : 用于生成Kafka主题的前缀。

例如，如果你只想捕获

inventory

customers

{
  "name": "mysql-inventory-connector",
  "config": {
    "connector.class": "io.debezium.connector.mysql.MySqlConnector",
    "database.hostname": "mysql.example.com",
    "database.port": "3306",
    "database.user": "debezium",
    "database.password": "dbz",
    "database.server.id": "184054",
    "database.server.name": "inventory",
    "database.include.list": "inventory",
    "table.include.list": "inventory.customers",
    "database.history.kafka.bootstrap.servers": "kafka:9092",
    "database.history.kafka.topic": "schema-changes.inventory",
    "snapshot.mode": "initial"
  }
}

这个配置首先指定了连接器类型为

MySqlConnector

database.server.name

database.include.list

table.include.list

database.history.kafka.bootstrap.servers

database.history.kafka.topic

snapshot.mode

initial

如何处理Debezium捕获的变更事件？

Debezium捕获的变更事件以JSON格式发布到Kafka主题。每个事件包含

before

after

source

op

• before

  : 变更前的数据，如果操作是插入，则为

  null

  。

• after

  : 变更后的数据，如果操作是删除，则为

  null

  。

• source

  : 包含关于变更事件来源的信息，如数据库名称、表名称、时间戳等。

• op

  : 表示操作类型，例如

  c

  （创建）、

  u

  （更新）、

  d

  （删除）、

  r

  （快照读取）。

下游应用需要解析这些JSON事件，并根据

op

op

c

after

op

u

op

d

使用Kafka Streams或Apache Flink等流处理框架可以方便地处理这些事件。例如，使用Kafka Streams可以这样处理：

KStream stream = builder.stream("inventory.customers");

stream.foreach((key, value) -> {
  try {
    JsonNode root = objectMapper.readTree(value);
    String op = root.get("payload").get("op").asText();

    if ("c".equals(op)) {
      JsonNode after = root.get("payload").get("after");
      // 将after中的数据插入到目标数据库
      System.out.println("Insert: " + after.toString());
    } else if ("u".equals(op)) {
      JsonNode after = root.get("payload").get("after");
      // 更新目标数据库中对应的数据
      System.out.println("Update: " + after.toString());
    } else if ("d".equals(op)) {
      JsonNode before = root.get("payload").get("before");
      // 从目标数据库中删除对应的数据
      System.out.println("Delete: " + before.toString());
    }
  } catch (Exception e) {
    e.printStackTrace();
  }
});

这段代码从

inventory.customers

op

如何监控和管理Debezium Connector？

监控和管理Debezium Connector对于确保数据同步的稳定性和可靠性至关重要。Kafka Connect提供了REST API，可以用于监控Connector的状态、配置和任务。

可以使用以下API来获取Connector的状态：

curl -X GET http://localhost:8083/connectors/mysql-inventory-connector/status

这个API会返回Connector的状态信息，包括状态（running、failed等）、任务状态以及错误信息。

还可以使用以下API来更新Connector的配置：

curl -X PUT \
  http://localhost:8083/connectors/mysql-inventory-connector/config \
  -H 'Content-Type: application/json' \
  -d '{
    "connector.class": "io.debezium.connector.mysql.MySqlConnector",
    "database.hostname": "mysql.example.com",
    "database.port": "3306",
    "database.user": "debezium",
    "database.password": "dbz",
    "database.server.id": "184054",
    "database.server.name": "inventory",
    "database.include.list": "inventory",
    "table.include.list": "inventory.customers",
    "database.history.kafka.bootstrap.servers": "kafka:9092",
    "database.history.kafka.topic": "schema-changes.inventory",
    "snapshot.mode": "never"
  }'

这个API会更新Connector的配置，例如修改

snapshot.mode

never

除了Kafka Connect API，还可以使用Debezium提供的JMX指标来监控Connector的性能。这些指标包括捕获的事件数量、延迟、错误率等。

微信 WeLM

WeLM不是一个直接的对话机器人，而是一个补全用户输入信息的生成模型。

下载

如果Connector出现问题，例如无法连接到MySQL或无法解析binlog事件，可以查看Kafka Connect的日志来排查问题。

如何处理Debezium Connector的Schema演化？

Schema演化是CDC过程中常见的问题。当MySQL表的结构发生变化时，例如添加、删除或修改列，Debezium需要能够正确处理这些变化。

Debezium通过Schema History Topic来管理Schema演化。每次表的结构发生变化时，Debezium会将新的Schema信息写入到Schema History Topic。下游应用可以读取Schema History Topic，并根据新的Schema来解析变更事件。

为了处理Schema演化，可以使用Avro或Protobuf等Schema注册表。这些注册表可以存储Schema信息，并为每个Schema分配一个唯一的ID。Debezium可以将Schema ID写入到变更事件中，下游应用可以使用Schema ID从注册表中获取Schema信息。

例如，使用Confluent Schema Registry可以这样配置Debezium Connector：

{
  "name": "mysql-inventory-connector",
  "config": {
    "connector.class": "io.debezium.connector.mysql.MySqlConnector",
    "database.hostname": "mysql.example.com",
    "database.port": "3306",
    "database.user": "debezium",
    "database.password": "dbz",
    "database.server.id": "184054",
    "database.server.name": "inventory",
    "database.include.list": "inventory",
    "table.include.list": "inventory.customers",
    "database.history.kafka.bootstrap.servers": "kafka:9092",
    "database.history.kafka.topic": "schema-changes.inventory",
    "snapshot.mode": "initial",
    "key.converter": "io.confluent.connect.avro.AvroConverter",
    "key.converter.schema.registry.url": "http://schema-registry:8081",
    "value.converter": "io.confluent.connect.avro.AvroConverter",
    "value.converter.schema.registry.url": "http://schema-registry:8081"
  }
}

这个配置指定了使用AvroConverter作为Key和Value的转换器，并配置了Schema Registry的URL。Debezium会将Avro Schema信息写入到Schema Registry，并将Schema ID写入到变更事件中。

下游应用可以使用Confluent提供的AvroDeserializer来读取变更事件，并从Schema Registry中获取Schema信息。

如何处理Debezium Connector的初始快照？

Debezium Connector在首次启动时会执行初始快照，将MySQL数据库中的所有数据读取到Kafka主题。初始快照可能会对MySQL数据库造成性能影响，特别是对于大型数据库。

为了减少初始快照对MySQL数据库的影响，可以采取以下措施：

• 使用

  snapshot.mode

  配置来控制快照模式。例如，可以设置为

  schema_only

  ，只读取表的结构，不读取数据；或者设置为

  never

  ，不执行快照。
• 使用

  snapshot.locking.mode

  配置来控制快照期间的锁模式。例如，可以设置为

  minimal

  ，使用最小的锁，减少对MySQL数据库的影响。
• 使用

  snapshot.new.tables

  配置来控制是否对新创建的表执行快照。
• 在MySQL数据库的低峰期执行初始快照。

如果初始快照失败，可以查看Kafka Connect的日志来排查问题。常见的错误包括无法连接到MySQL、权限不足或内存不足。

如何优化Debezium Connector的性能？

优化Debezium Connector的性能可以提高数据同步的速度和可靠性。以下是一些优化建议：

• 增加Kafka Connect集群的资源，例如CPU和内存。
• 调整Kafka Connect的配置，例如

  tasks.max

  和

  consumer.override.max.poll.records

  。
• 优化MySQL数据库的配置，例如

  binlog_format

  和

  binlog_row_image

  。
• 使用合适的Schema注册表，例如Confluent Schema Registry。
• 监控Debezium Connector的性能指标，例如捕获的事件数量、延迟和错误率。

如何确保Debezium Connector的数据一致性？

确保Debezium Connector的数据一致性是CDC的关键目标。以下是一些建议：

• 使用事务性Outbox模式来确保数据变更的原子性。
• 使用Debezium提供的Heartbeat功能来检测数据同步的延迟。
• 使用Kafka的事务性功能来确保数据同步的Exactly-Once语义。
• 定期验证目标数据库中的数据与MySQL数据库中的数据是否一致。

总的来说，使用Debezium进行MySQL CDC需要仔细规划和配置，并根据实际情况进行优化。通过合理的配置和监控，可以实现高效、可靠的数据同步，为微服务架构和数据集成提供强大的支持。