Connect to OceanBase Database by using Debezium and obtain data

Debezium is an open-source distributed platform that monitors and captures database changes and turns them into event streams. Debezium is implemented based on Apache Kafka and supports multiple types of database systems.

Prerequisites

You have installed oblogproxy.

Examples

In this topic, a docker is used to demonstrate how to connect to OceanBase Database by using Debezium and obtain data. This deployment mode cannot be directly used in a production environment. Perform the following steps:

1. Start Zookeeper.

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

2. Start Kafka.

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

3. Start Kafka Connect, which exposes a REST API to manage the Debezium connector for 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. Create a file named payload.json and write the following sample code to the file:

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