---
title: "使用 Debezium 连接 OceanBase 并获取数据 - OceanBase 数据库 V4.3.1 | OceanBase 文档中心"
description: 使用 Debezium 连接 OceanBase 并获取数据 Debezium 是一个开源的分布式平台，用于监控数据库变化和捕捉数据变动事件，并以事件流的形式导出到各种消费者。Debezium 基于 Apache Kafka 实现，并支持多种数据库系统。 前提条件 在使用 Debezium 之前，确认您已安装 obl…
---
切换语言

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

文档反馈![](https://mdn.alipayobjects.com/huamei_22khvb/afts/img/A*P8CuR4UJ_FkAAAAAAAAAAAAADiGDAQ/original) OceanBase 数据库分布式版 - V 4.3.1

# 使用 Debezium 连接 OceanBase 并获取数据

更新时间：2026-04-10 11:58:01

[编辑](https://github.com/oceanbase/oceanbase-doc/edit/V4.3.1/zh-CN/680.ecological-integration/400.data-ingestion/1600.debezium.md)  

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

## 前提条件

在使用 Debezium 之前，确认您已安装 oblogproxy。更多信息，参考 [安装 oblogproxy 文档](https://www.oceanbase.com/docs/community-oblogproxy-doc-1000000000438540)。

## 使用示例

本文以 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` 参数过滤。
 3. 使用curl命令向 Kafka Connect 服务的 API 提交一个 POST 请求到 /connectors 资源，并附带上述新连接器的 JSON 文档。  
   ` cat payload.json | tr -d "\n" | curl -X POST -H "Content-Type: application/json" -d @- localhost:8083/connectors/`
 4. 注册一个连接器之后，它会在 Kafka Connect 容器中生成大量的日志输出。从这些输出，你可以观察到连接器从创建到开始读取 OceanBase 的 Binlog 所经历的过程。
   `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) 咨询热线