EventBridge:Debezium MySQL Source コネクタを使用した MySQL から ApsaraMQ for Kafka へのデータ同期 (非推奨)

仕組み

Debezium MySQL コネクタは MySQL のバイナリログ (binlog) を読み取り、各行レベルの変更を構造化されたイベントに変換します。Serverless App Engine (SAE) 上でマネージドサービスとして実行される Kafka Connect は、これらのイベントを ApsaraMQ for Kafka の送信先トピックに発行します。

データは 3 つのステージを通過します：

1. MySQL は、コミットされたすべてのトランザクションを binlog に記録します。

2. Debezium コネクタは binlog を読み取り、変更イベントを生成します。

3. Kafka Connect は、各イベントを ApsaraMQ for Kafka の送信先トピックに発行します。

初回起動時、コネクタは既存データの初期スナップショットを取得し、その後、継続的な変更のために binlog の読み取りに切り替わります。

制限事項

• Debezium MySQL コネクタは、1 つのタスク (tasks.max = 1) のみをサポートします。複数のワーカーでフォールトトレランスを提供することはできますが、タスク自体を並列化することはできません。

• コネクタのバージョンは Java 8 と互換性がある必要があります。

前提条件

開始する前に、以下のサービスとリソースが準備できていることを確認してください：

• EventBridge が有効化され、Resource Access Management (RAM) 権限が付与されていること

• Object Storage Service (OSS) が有効化され、バケットが作成されていること

• Serverless App Engine (SAE) が有効化されていること

• Virtual Private Cloud (VPC) と vSwitch が作成されていること

• ApsaraMQ for Kafka インスタンスが購入され、デプロイされていること

ステップ 1：MySQL ソーステーブルの作成

ApsaraDB RDS for MySQL インスタンスをセットアップし、コネクタが監視するテーブルを作成します。

1. ApsaraDB RDS コンソールにログインし、ApsaraDB RDS for MySQL インスタンスを作成します。

   重要

   ApsaraMQ for Kafka インスタンスと同じ VPC を選択し、VPC CIDR ブロックをインスタンスのホワイトリストに追加してください。

   

2. [インスタンス] ページで、作成したインスタンスをクリックします。インスタンスの詳細ページで、以下の操作を完了します。

   1. 左側のナビゲーションウィンドウで、[アカウント] をクリックし、「アカウントを作成」をご参照ください。また、既存のアカウントを使用することもできます。

   2. 左側のナビゲーションウィンドウで、[データベース] をクリックし、データベースを作成することができます。既存のデータベースを使用することもできます。

   3. 左側のナビゲーションウィンドウで、[データベース接続] をクリックし、内部エンドポイント と ポート番号 をメモします。コネクタを設定する際に、これらの値が必要になります。

   

3. 左側のナビゲーションウィンドウで、[基本情報] をクリックし、[データベースにログオン] をクリックして Data Management (DMS) コンソールを開きます。

   1. DMS の左側のナビゲーションウィンドウで、データベース名をダブルクリックして切り替えます。

      

   2. [SQLコンソール] タブで、以下の SQL ステートメントを実行してサンプルテーブルを作成します：

      CREATE TABLE sql_table(id INT, number INT);

ステップ 2：Kafka Connect コネクタのセットアップ

Debezium プラグインをダウンロードし、OSS にアップロードして、ApsaraMQ for Kafka コンソールでコネクタタスクを設定します。

コネクタの検証

テストレコードを挿入し、変更イベントが Kafka に表示されるか確認します。

1. DMS コンソールで、ステップ 1 で作成したテーブルにレコードを挿入します：

   INSERT INTO sql_table(id, number) VALUES(123, 20000);

2. 「ApsaraMQ for Kafka コンソール」にログインします。[インスタンス] ページで、使用中のインスタンスをクリックします。

3. 左側のナビゲーションウィンドウで、[トピック] をクリックし、目的のトピックを選択して、[メッセージクエリ] タブを開きます。

4. 各トピックのレコードと期待されるメッセージフォーマットを確認します：

   データベースレベルのトピック ({database.server.name}) -- スキーマ変更イベントが含まれます：

   {
       "source": {
           "version": "1.5.0.Final",
           "connector": "mysql",
           "name": "fulfillment",
           "ts_ms": 1686283675404,
           "snapshot": "true",
           "db": "eb-ceshi",
           "table": "sql_table",
           "server_id": 0,
           "file": "mysql-bin.000006",
           "pos": 188032,
           "row": 0
       },
       "databaseName": "eb-cesh",
       "ddl": "DROP TABLE IF EXISTS sql_table",
       "tableChanges": []
   }

   テーブルレベルのトピック ({database.server.name}.{databaseName}.{tableName}) -- 変更前後のデータを含む行変更イベントが含まれます：

   {
       "before": null,
       "after": {
           "id": 123,
           "number": 20000
       },
       "source": {
           "version": "1.5.0.Final",
           "connector": "mysql",
           "name": "fulfillment",
           "ts_ms": 1686283675675,
           "snapshot": "last",
           "db": "eb-cesh",
           "table": "sql_table",
           "server_id": 0,
           "file": "mysql-bin.000006",
           "pos": 188032,
           "row": 0
       },
       "op": "r",
       "ts_ms": 1686283675675,
       "transaction": null
   }

   スキーマ履歴トピック (database.history.kafka.topic) -- DDL 変更レコードが含まれます：

   {
       "source": {
           "server": "fulfillment"
       },
       "position": {
           "ts_sec": 1686283675,
           "file": "mysql-bin.000006",
           "pos": 188032,
           "snapshot": true
       },
       "databaseName": "eb-cesh",
       "ddl": "CREATE DATABASE `wbdb` CHARSET utf8 COLLATE utf8_general_ci",
       "tableChanges": []
   }

トラブルシューティング

All tasks under connector mongo-source failed, please check the error trace of the task.

Kafka Connect が終了しました！SAE アプリケーションのエラーログ /opt/kafka/logs/connect.log を確認して、Kafka Connect が終了した原因を特定し、有効な引数でイベントストリームを更新して問題を解決してください。

Start or update connector xxx failed. Error code=400. Error message=Connector configuration is invalid and contains the following 1 error(s):
Value must be one of never, initial_only, when_needed, initial, schema_only, schema_only_recovery
You can also find the above list of errors at the endpoint `/connector-plugins/{connectorType}/config/validate`

curl -i -X PUT -H "Accept:application/json" \
  -H "Content-Type:application/json" \
  -d @$CONNECTOR_PROPERTIES_MAPPING \
  http://localhost:8083/connector-plugins/io.confluent.connect.jdbc.JdbcSinkConnector/config/validate

{
    "value": {
        "name": "snapshot.mode",
        "value": null,
        "recommended_values": [
            "never",
            "initial_only",
            "when_needed",
            "initial",
            "schema_only",
            "schema_only_recovery"
        ],
        "errors": [
            "Value must be one of never, initial_only, when_needed, initial, schema_only, schema_only_recovery"
        ],
        "visible": true
    }
}

org.apache.kafka.common.errors.TimeoutException: Timeout expired while fetching topic metadata

参考文献

• Debezium MySQL コネクタのプロパティ

• EventBridge を有効化して権限を付与する

• 柔軟なグループ作成機能の使用