---
title: "Debezium - OB Cloud 云数据库 master | OceanBase 文档中心"
description: Debezium Debezium 是一个开源的分布式平台，用于监控数据库变化和捕捉数据变动事件，并以事件流的形式导出到各种消费者。Debezium 基于 Apache Kafka 实现，并支持多种数据库系统。 前提条件 开始操作前，开启 OB Cloud 的 Binlog 服务。 开启 Binlog 日志服务的路径…
---
切换语言

- 中文站 - 简体中文
- International - English
- 日本站 - 日本語

文档反馈![](https://mdn.alipayobjects.com/huamei_22khvb/afts/img/A*qbZXRo_94ZEAAAAAAAAAAAAADiGDAQ/original) OB Cloud 云数据库

# Debezium

更新时间：2026-06-09 15:42:58

Debezium 是一个开源的分布式平台，用于监控数据库变化和捕捉数据变动事件，并以事件流的形式导出到各种消费者。Debezium 基于 Apache Kafka 实现，并支持多种数据库系统。

## 前提条件

开始操作前，开启 OB Cloud 的 Binlog 服务。

开启 Binlog 日志服务的路径：实例列表 -> 租户管理 -> Binlog 服务，点击开通即可。更多信息，参考 [开通 Binlog 日志服务](https://www.oceanbase.com/docs/common-oceanbase-cloud-1000000000369157#1-title-%E5%BC%80%E9%80%9A%20Binlog%20%E6%97%A5%E5%BF%97%E6%9C%8D%E5%8A%A1)。

#### 注意

1. 不支持 Initial snapshots， 只支持默认的 Incremental snapshots。

2. 建议将 snapshot.locking.mode 设为 `none`，且阅读官方文档的风险提示。

## 使用示例

本文以 Docker 的方式简单示范如何使用 Debezium 连接 OceanBase 并获取数据，本部署方式不适合直接用于生产环境。按照以下步骤操作：

1. 启动 Zookeeper

   ```shell
   docker run -it -d --name zookeeper -p 2181:2181 -p 2888:2888 -p 3888:3888 quay.io/debezium/zookeeper:2.4

   ```
 2. 启动 Kafka

   ```shell
   docker run -it -d --name kafka -p 9092:9092 --link zookeeper:zookeeper quay.io/debezium/kafka:2.4

   ```
 3. 启动 Kafka Connect 服务。该服务暴露了一个 REST API 用于管理 Debezium MySQL 连接器。

   ```shell
   docker run -it -d --name connect -p 8083:8083 -e GROUP_ID=1 -e CONFIG_STORAGE_TOPIC=my_connect_configs -e OFFSET_STORAGE_TOPIC=my_connect_offsets -e STATUS_STORAGE_TOPIC=my_connect_statuses --link kafka:kafka  quay.io/debezium/connect:2.4

   ```
 4. 新建 `payload.json` 文件，写入以下示例配置。

   ```json
      {
      "name": "inventory-connector",
      "config": {
        "connector.class": "io.debezium.connector.mysql.MySqlConnector",
        "tasks.max": "1",
        "database.hostname": "xxxx.cn-hangzhou.oceanbase.aliyuncs.com",
        "database.port": "3306",
        "database.user": "root",
        "database.password": "xxxx",
        "database.server.id": "1",
        "topic.prefix": "observer1",
        "database.include.list": "debe",
        "table.include.list":"debe.earthquake",
        "schema.history.internal.store.only.captured.tables.ddl":true,
        "schema.history.internal.skip.unparseable.ddl":true,
        "schema.history.internal.kafka.bootstrap.servers": "kafka:9092",
        "schema.history.internal.kafka.topic": "schema-changes.inventory",
        "snapshot.locking.mode": "none"
      }
    }

   ```

   #### 注意

      1. MySQL 连接器无法识别 OceanBase 的 `PURGE TABLE` 等语法，可以配置 `schema.history.internal.skip.unparseable.ddl` 选项规避。
      2. 您可能并不希望同步 OceanBase 独有的虚拟表和视图，可以使用 `table.include.list 或者 table.exclude.list` 参数过滤。
 5. 使用 `curl` 命令向 Kafka Connect 服务的 API 提交一个 POST 请求到 `/connectors` 资源，并附带上述新连接器的 JSON 文档。

   ```shell
   	cat payload.json | tr -d "\n" | curl -X POST -H "Content-Type: application/json" -d @- localhost:8083/connectors/

   ```
 6. 注册一个连接器之后，它会在 Kafka Connect 容器中生成大量的日志输出。从这些输出，你可以观察到连接器从创建到开始读取 OceanBase 的 Binlog 所经历的过程。

   ```shell
   	docker run -it --rm --name watcher --link zookeeper:zookeeper --link kafka:kafka quay.io/debezium/kafka:2.4 watch-topic -a -k observer1.debe.earthquake

   ```

```

```

 上一篇 下一篇 ![有帮助](https://gw.alipayobjects.com/mdn/ob_asset/afts/img/A*y6ocSqN8cqsAAAAAAAAAAAAAARQnAQ)![无帮助](https://gw.alipayobjects.com/mdn/ob_asset/afts/img/A*BG9IQJyLHF8AAAAAAAAAAAAAARQnAQ)![反馈](https://gw.alipayobjects.com/mdn/ob_asset/afts/img/A*eTWdQKCRKHwAAAAAAAAAAAAAARQnAQ)[AI](https://www.oceanbase.com/obi) 咨询热线