使用 Debezium 连接 OceanBase 并获取数据

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

前提条件

在使用 Debezium 之前，确认您已安装 oblogproxy。更多信息，参考 安装 oblogproxy 文档。

使用示例

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

1. 启动 Zookeeper

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

2. 启动 Kafka

   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 连接器。

   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 文件，写入以下示例配置。

      {
      "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"
      }
    }