OceanBaseデータベースとKafkaの統合に関するベストプラクティス|OceanBase Cloud|ドキュメント|OceanBaseマニュアル

Kafkaは高性能な分散ストリーム処理プラットフォームです。通常、リアルタイムデータストリームの処理に使用され、リアルタイムデータパイプライン、ストリーム処理アプリケーション、リアルタイム分析システムなどのシナリオで広く利用されています。

KafkaおよびKafka Connectモジュールを使用し、サードパーティのコネクタを導入することで同様の機能を実現できます。現在、Debezium公式ではOceanBaseデータベース専用のコネクタはありませんが、OceanBaseデータベースはMySQLと互換性があるため、本記事ではDebezium MySQL ConnectorとConfluent JDBC Sink ConnectorをKafkaとOceanBaseデータベースの中間橋渡し役として使用し、OceanBaseデータベースとKafkaのデータ統合を行います。

背景

Debeziumは、Kafka Connect用に開発された一連のソースコネクタであり、ログをキャプチャしてデータベースの変更を取得します。分散サービスとして、Debeziumは各データテーブルの行レベルの変更を記録し、Kafkaのトピックを通じてアプリケーションにデータ変更を提供します。初期化時に、既存のすべてのデータをKafkaトピックに送信し、新しいデータ変更をキャプチャすることができます。

Confluent JDBC Sink Connectorは、Kafkaデータをリレーショナルデータベースに書き込むためのKafkaコネクタです。ユーザーは、Kafkaトピック内のメッセージをJDBC互換データベースにストリーミング配信することで、データのリアルタイム同期と永続化ストレージを実現できます。

Kafka Connectのワークフロー

ソースコネクタを設定します。本記事ではDebezium MySQL Connectorを使用し、OceanBaseデータベースからデータを読み取ります。
読み取ったデータをKafka内の特定のトピックに書き込みます。
Sinkコネクタを設定します。本記事ではConfluent JDBC Sink Connectorを使用し、Kafka内の特定のトピックからデータを読み取り、OceanBaseデータベースに書き込みます。

デモ環境の紹介

OceanBaseデータベースとKafkaの統合を開始する前に、以下を確認してください：

DebeziumのバージョンがV1.5.4.Finalであること。
KafkaのバージョンがV2.12からV2.5.0であること。
ZookeeperのバージョンがV3.6であること。
OceanBaseデータベースのバージョンがV4.2.3であり、Binlogが有効になっていること。
Confluent JDBC Sink ConnectorのバージョンがV10.7.6であること。

説明

本記事のデモ環境のバージョンは参考用です。他のバージョンでも構いませんが、互換性を確保してください。

OceanBaseデータベースの設定

以下の手順に従ってOceanBaseデータベースを設定します：

OceanBase CloudのBinlogサービスを有効にします。 Binlogサービスを有効にするパス：インスタンスリスト -> テナント管理 -> Binlogサービス、有効にするをクリックします。詳細については、Binlogサービスの有効化をご参照ください。
OceanBaseデータベースに接続し、Debezium Connector専用のアカウントを作成します。
```
CREATE USER 'debezium_user'@'localhost' IDENTIFIED BY 'debezium_password';
```
OceanBase Cloudを使用している場合は、OceanBaseクラウドサービスコンソールにログインしてユーザーを作成し、権限を付与します。詳細については、アカウント（データベースユーザー）の作成をご参照ください。

debezium_user アカウントに権限を付与します。

GRANT SELECT, CREATE, RELOAD, SHOW DATABASES ON *.* TO 'debezium_user' IDENTIFIED BY 'debezium_password';

OceanBaseデータベースでBinlogサービスを有効にします。

OceanBase Cloudを使用している場合は、以下の方法でBinlogを有効にします：

Binlogサービスを有効にするパス：インスタンスリスト -> テナント管理 -> Binlogサービス、有効にするをクリックします。詳細については、Binlogサービスの有効化をご参照ください。

以下のコマンドを実行して、Binlogが有効になっていることを確認します：

SHOW MASTER STATUS;

期待される戻り値：

+------------------+----------+--------------+------------------+------------------------------------------+
| File             | Position | Binlog_Do_DB | Binlog_Ignore_DB | Executed_Gtid_Set                        |
+------------------+----------+--------------+------------------+------------------------------------------+
| mysql-bin.000001 |     2567 |              |                  | a2750d9c-11da-11ef-81aa-0242ac110008:1-9 |
+------------------+----------+--------------+------------------+------------------------------------------+
1 row in set (0.125 sec)

interactive_timeout と wait_timeout のパラメータ値を確認します。
```
SHOW VARIABLES LIKE 'interactive_timeout';   
SHOW VARIABLES LIKE 'wait_timeout';
```
これら2つのパラメータのデフォルト値はどちらも28800秒、つまり8時間です。初期化スナップショットを実行する際、時間がかかりすぎて接続がタイムアウトする可能性があります。実際の状況に応じて、これら2つのパラメータを調整するかどうかを決定できます。

Debezium MySQL Connectorの設定

以下の手順に従って、Debezium MySQL Connectorを設定します：

Kafka Connectがインストールされているマシンに、Debezium MySQL Connector plug-in 1.5.4.Finalバージョンをダウンロードします。

詳細については、Debezium MySQL connectorのダウンロードURLをご参照ください。
Debezium MySQL Connectorを解凍します。

OceanBaseデータベースからKafkaへのデータ送信

このセクションでは、KafkaとDebeziumを設定し、OceanBaseデータベースのデータを同期します。

Debezium MySQL Connectorの起動

以下の手順に従って、分散型でDebezium MySQL Connectorを起動します：

Kafka Connectの設定

Kafkaのインストールディレクトリ配下のconfigフォルダに移動します。この例では、ディレクトリは/opt/kafka/configです。connect-distributed.propertiesファイルを編集し、ファイルの最終行にDebezium MySQL connectorプラグインのパスを追加します。
```
# 設定ファイルを開く
vi /opt/kafka/config/connect-distributed.properties
# 以下の内容を追加：
plugin.path=/opt/debezium/plugin
```
注意

パスにはplugin配下のdebezium-connector-mysqlディレクトリを含めないでください。
Kafka Topicsの状態を確認します。

Kafka Connectを起動する前に、現在のトピックの状態を確認します。
```
bin/kafka-topics.sh --list --bootstrap-server localhost:9092
```
本文中は現在空で、トピックはありません。
Kafka Connectを起動します。

以下のコマンドを使用してKafka Connectを起動します：
```
bin/connect-distributed.sh -daemon config/connect-distributed.properties
```
OceanBaseデータベースの同期情報を設定します。

任意のディレクトリにregister-oceanbase-debezium.jsonファイルを作成し、OceanBaseデータベースの同期に関する情報を設定します。
```
# register-oceanbase-debezium.jsonファイルを作成
vi register-oceanbase-debezium.json
```
以下の内容を追加します：
```
{
    "name": "oceanbase-inventory-connector",
    "config": {
        "connector.class": "io.debezium.connector.mysql.MySqlConnector",
        "tasks.max": "1",
        "database.hostname": "xxx.xx.x.x",
        "database.port": "xxxx",
        "database.user": "debezium_user",
        "database.password": "xxxxxxxx",
        "database.server.id": "1",
        "database.server.name": "zhang-oceanbase",
        "database.whitelist": "test",
        "database.history.kafka.bootstrap.servers": "localhost:9092",
        "database.history.kafka.topic": "oceanbase-schema-changes-inventory",
        "snapshot.locking.mode": "none"
    }
}
```
詳細なパラメータ情報については、Debezium MySQL Connectorドキュメントをご参照ください。

説明

上記の設定では、database.server.name は zhang-oceanbase、database.whitelist は test です。Debeziumは、許可リスト内のデータベースの各テーブルに対して、{database.server.name}.{データベース名}.{テーブル名} の形式でトピックを自動作成します。したがって、OceanBaseのtestデータベースにテーブルtb1が存在する場合、トピックzhang-oceanbase.test.tb1が自動生成されます。testデータベースに同期が必要なテーブルが既に作成されていることを確認してください。以下の例では、テーブルtb1を例に説明します。
注意
- パラメータをご自身のデータベースの関連情報に変更してください。
- 初期スナップショット段階では、他のクライアントがテーブル構造を変更しないことを確信している場合に限り、snapshot.locking.mode を none に設定することが安全です。
REST APIを使用してDebezium MySQL Connectorを登録します。

以下のコマンドを使用して、設定をDebezium MySQL Connectorに登録します：
```
curl -X POST -H 'Content-Type: application/json' --data @register-oceanbase-debezium.json http://localhost:8083/connectors/
```
注意

register-oceanbase-debezium.json ファイルを含むパスでこのコマンドを実行する必要があります。
Connectorが正常に追加されたかどうかを確認します。
```
curl http://localhost:8083/connectors
```
成功すれば、Connectorのリストが表示されます。
Kafkaトピックを確認します。
```
bin/kafka-topics.sh --list --bootstrap-server localhost:9092
```
この時点で、関連するトピックのリストが表示されるはずです。
Kafkaトピック内のデータメッセージを確認します。

以下のコマンドを実行し、データがKafkaトピックに送信されたことを確認します：
```
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic zhang-oceanbase.test.tb1 --from-beginning
```
この例では、test データベースにテーブル tb1 が含まれている場合、14件のデータが存在し、ID の最大値は14になります。

上記の手順では、OceanBaseのデータをKafkaに同期し、Debeziumが自動的に作成したトピック（この例の zhang-oceanbase.test.tb1 など）にデータを送信する方法を示しました。

データの追加

データを1件追加し、新規データがKafkaのトピックzhang-oceanbase.test.tb1に送信されているか確認します。

INSERT INTO tb1 (id,name,addtime) VALUES (15,'xiaoli','2024-06-18 17:20:00');

コネクタの更新と削除

コネクタ設定を更新します。

更新と新規作成のJSONファイル形式は異なります。更新時にはnameとconfigの2つのキーがないため、更新時のファイル形式は次のとおりです：

{
    "connector.class": "io.debezium.connector.mysql.MySqlConnector",
    "tasks.max": "1",
    "database.hostname": "xxx.xx.x.x",
    "database.port": "xxxxx",
    "database.user": "debezium_user@zhang",
    "database.server.id": "1",
    "database.server.name": "zhang-oceanbase",
    "database.whitelist": "test",
    "database.history.kafka.bootstrap.servers": "localhost:9092",
    "database.history.kafka.topic": "oceanbase-schema-changes-inventory",
    "snapshot.locking.mode": "none"
}

REST APIを使用して更新します。メソッドはPUTです：

curl -X PUT -H 'Content-Type: application/json' --data @register-oceanbase-update.json http://localhost:8083/connectors/oceanbase-inventory-connector/config

コネクタを削除します。
```
curl -X DELETE http://localhost:8083/connectors/oceanbase-inventory-connector
```
コネクタを削除しても、Kafkaのトピックは削除されません。
同一のコネクタを再作成します。

削除後、同一のコネクタを再作成します。新規データは増分形式で、引き続き前回作成したトピックに送信されます。

KafkaのメッセージをOceanBaseデータベースに書き込む

以下の手順に従い、JDBC Sinkを使用してKafkaのTopicからデータを消費します：

Confluent Kafka Connect JDBCを準備します。
1. Confluent Kafka Connect JDBCをダウンロードします。詳細については、JDBC Connector (Source and Sink)をご参照ください。デプロイ方法はSelf-Hostedを選択します。
2. ダウンロードした圧縮ファイルを解凍し、解凍後のフォルダをconnect-distributed.propertiesファイル内のplugin.path設定のパスに配置します。
3. libディレクトリにMySQLドライバがないため、MySQLドライバをlibディレクトリに配置します。以下のコマンドを実行して、/opt/debezium/plugin/debezium-connector-mysql配下のドライバをコピーします：
```
cp /opt/debezium/plugin/debezium-connector-mysql/mysql-connector-java-8.0.21.jar /opt/debezium/plugin/confluentinc-kafka-connect-jdbc-10.7.6/lib/
```
Kafka Connectを再起動します。
1. jpsコマンドを使用してConnectDistributedプロセスを見つけ、kill pidを実行してからKafka Connectを再起動します。
```
bin/connect-distributed.sh -daemon ./config/connect-distributed.properties
```
2. Kafka Connectを再起動した後、JDBC Sink Connectorが正常に追加されたかどうか確認します：
```
curl http://localhost:8083/connector-plugins
```
  期待される戻り値：
```
{"class":"io.confluent.connect.jdbc.JdbcSinkConnector","type":"sink","version":"10.7.6"}
```

Kafka Connect JDBC Sinkの設定ファイルを作成します。

vi register-oceanbase-sink.json

register-oceanbase-sink.jsonファイルに以下の内容を追加します：

{
    "name": "connect-oceanbase-sink",
    "config": {
        "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
        "tasks.max": "1",
        "connection.url": "jdbc:mysql://xxx.xx.x.x:xxxx/test_sink?useUnicode=true&characterEncoding=UTF-8&useSSL=false",
        "connection.user": "xxxxx",
        "connection.password": "xxxxxx",
        "insert.mode": "upsert",
        "delete.enabled": "true",
        "pk.mode": "record_key",
        "auto.create": "true",
        "auto.evolve": "true",
        "topics": "zhang-oceanbase.test.tb1",
        "table.name.format": "tb2",
        "transforms": "ExtractField",
        "transforms.ExtractField.type": "org.apache.kafka.connect.transforms.ExtractField$Value",
        "transforms.ExtractField.field": "after"
    }
}

注意

例ではSink専用のユーザーを作成せず、rootユーザーを使用しています。実際の運用では状況に応じて選択できます。
ソースデータベースとターゲットデータベースは異なるテナントに配置されています。同一テナントの場合、Binlog記録に問題が発生する可能性があります。test_sinkは事前に作成済みのターゲットデータベースです。

Kafka ConnectにJDBC Sink Connectorを追加します。

curl -X POST -H "Content-Type: application/json" --data @register-oceanbase-sink.json http://localhost:8083/connectors

JDBC Sink Connectorが正常に追加されたかどうか確認します：

curl http://localhost:8083/connectors

期待される実行結果：

["connect-oceanbase-sink","oceanbase-inventory-connector"]

フルデータの同期

test_sinkに移動してデータを確認します：

SELECT * FROM tb2;

期待される実行結果：

+----+-----------+----------------------+
| ID | name      | addtime              |
+----+-----------+----------------------+
|  1 | xiaozhao  | 2016-12-09T16:04:33Z |
|  2 | xiaozhang | 2016-12-09T16:04:33Z |
|  3 | xiaozhang | 2016-12-09T16:04:33Z |
|  4 | xiaozhang | 2017-12-09T16:04:33Z |
|  5 | xiaozhang | 2017-12-09T16:04:33Z |
|  6 | xiaozhang | 2017-12-09T16:04:33Z |
|  7 | xiaozhang | 2017-12-09T16:04:33Z |
|  8 | xiaozhang | 2017-12-09T16:04:33Z |
|  9 | xiaozhang | 2017-12-09T16:04:33Z |
| 10 | xiaozhang | 2017-12-09T16:04:33Z |
| 11 | xiaozhang | 2017-12-09T16:04:33Z |
| 12 | xiaozhang | 2017-12-09T16:04:33Z |
| 13 | xiaozhang | 2017-12-09T16:04:33Z |
| 14 | xiaozhang | 2017-12-09T16:04:33Z |
| 15 | xiaoli    | 2024-06-18T09:20:00Z |
| 16 | xiaoli    | 2024-06-18T09:20:00Z |
| 17 | xiaoli    | 2024-06-18T09:20:00Z |
| 18 | xiaoli    | 2024-06-18T09:20:00Z |
| 19 | xiaoli    | 2024-06-18T09:20:00Z |
| 20 | xiaoli    | 2024-06-18T09:20:00Z |
| 21 | xiaoli    | 2024-06-18T09:20:00Z |
| 22 | xiaoli    | 2024-06-18T09:20:00Z |
| 23 | xiaoli    | 2024-06-18T09:20:00Z |
| 24 | xiaozhang | 2024-06-24T09:20:00Z |
+----+-----------+----------------------+
24 rows in set (0.013 sec)

増分データの同期
1. ソースデータベース test にIDが25のデータを1件追加します：
```
INSERT INTO tb1 (id, name, addtime) VALUES (25,'xiaozhang','2024-06-25 09:20:00');
```
2. ターゲットデータベース test_sink で、IDが25の増分データを確認します：
```
SELECT * FROM tb2 WHERE id = 25;
```
テーブル構造の変更

ソーステーブルに age 列を追加します：
```
ALTER TABLE tb1 ADD COLUMN age INT;
```
この時点で、元のテーブルには1列が追加されましたが、ターゲットテーブルは変更されていません。これは、新しいデータがターゲットテーブルに到達していないためです。ターゲットテーブルの構造は、新規データが追加された場合にのみ変更されます。
ソーステーブルにIDが26のデータをもう1件追加します：
```
INSERT INTO tb1 (id, name, addtime, age) VALUES (26,'xiaozhang','2024-06-25 10:20:00',20);
```
ターゲットテーブルの構造が変更され、IDが26のデータがターゲットテーブルに同期されたことが確認できます。

テーブルデータの更新

ターゲットテーブルのIDが26の行の age が21に変更されました。

UPDATE tb1 SET age = 21 WHERE id = 26;

テーブルデータの削除

ソーステーブルからIDが20のデータを削除します：
```
DELETE FROM tb1 WHERE id = 20;
```
ターゲットテーブルの対応するデータが削除されています：
```
SELECT * FROM tb2 WHERE id = 20;
```