OceanBaseデータベースのデータをRocketMQに移行する

本記事では、データ移行サービスを使用して、OceanBaseデータベース（MySQL互換モードおよびOracle互換モードを含む）のデータをRocketMQに移行する方法について説明します。

背景

RocketMQは、高スループット、低遅延、高可用性、高信頼性を備えた分散型メッセージミドルウェアであり、取引、決済、注文、物流、通知、ログ収集、データ同期などの業務シナリオで広く利用されています。その中核となる機能は、異なるシステム間で非同期メッセージ通信を提供することです。プロデューサー側は業務イベントやデータ変更をRocketMQに書き込み、コンシューマー側は自身の処理能力に応じてRocketMQからメッセージをサブスクライブして消費することで、システムの疎結合、非同期処理、トラフィックのピークカット、およびイベント駆動アーキテクチャを実現します。

データ統合やリアルタイムデータ同期のシナリオでは、RocketMQは下流のメッセージシステムとして、データベース、業務システム、またはログシステムからの変更データを受け持つ役割を果たすことがよくあります。データ移行サービスは、OceanBaseデータベースの2種類の互換モードとRocketMQデータソース間のリアルタイムデータ同期をサポートしており、下流の業務システム、リアルタイム計算システム、データ処理サービス、または異種ストレージシステムへの提供が可能です。

セルフマネージド RocketMQを使用している場合、プライベートネットワーク接続方式による移行プロセスを理解する際には、以下のネットワーク接続タスクフローを参照してください。

前提条件

• ソース側のOceanBaseデータベースのクラスタとテナントが作成済みであること。詳細については、クラスタの作成およびテナントの作成を参照してください。

• ソース側とターゲット側に、データ移行専用のデータベースユーザーを作成し、関連する権限を付与していること。詳細については、ユーザー権限の説明を参照してください。

使用上の制限

• データ移行タスクを新規作成できるのは、プロジェクトロールが プロジェクト所有者、プロジェクト管理者、または データサービス管理者 のユーザーのみです。

• 現在、データ転送でサポートされているRocketMQインスタンスのバージョンはV4.xおよびV5.xであり、商用版とコミュニティ版を含みます。

• データ移行サービスは、データベース名、テーブル名、列名がASCIIコードであり、特殊文字（.|"'`()=;/& および改行を含む）を含まないオブジェクトの移行のみをサポートします。

• 移行対象のテーブル名および列名には、中国語の文字を含めることはできません。

• データ移行の対象として物理テーブルのみをサポートし、他のオブジェクトはサポートしません。

• データ移行プロセスにおいて、データ転送はテーブルを削除してから新規作成することをサポートしています。つまり、既に同期されているテーブルに対してDROP TABLE操作を実行した後、CREATE TABLEを実行することが可能です。データ転送は、リネーム操作によるテーブルの新規作成、すなわちRENAME TABLE a TO a_tmp操作の実行をサポートしません。

• 現在、通常のメッセージTopicの使用のみをサポートしています。

注意事項

• ノード間の時刻が同期されていない場合、またはクライアント端末とサーバー間の時刻が同期されていない場合、増分同期の遅延時間が正確でない可能性があります。

  例えば、時計が標準時より進んでいる場合、遅延時間が負の値になる可能性があります。時計が標準時より遅れている場合、遅延が発生する可能性があります。

• ソースがOceanBaseデータベースでDDL同期が有効なデータ移行タスクにおいて、ソース側のデータベーステーブルでRENAME操作が発生した場合、増分同期でデータが失われないよう、タスクを再起動することを推奨します。

• OceanBaseデータベースがV4.0.0からV4.3.xの間のバージョン（V4.2.5 BP1を除く）で、かつ増分同期が選択されている場合は、生成列にSTORED属性を設定してください。そうでない場合、増分ログには生成列の情報が保存されず、増分同期データに異常が発生する可能性があります。

• 更新される行にLOB列が含まれる場合：

  • LOB列が更新列である場合、UPDATEまたはDELETE操作前のLOB列の値に依存しないでください。

    現在、LOB列を使用して格納されているデータ型には、JSON、GIS、XML、UDT（ユーザー定義型）、およびLONGTEXT、MEDIUMTEXTなどの各種TEXTが含まれます。

  • LOB列が非更新列の場合、UPDATEまたはDELETE操作の前後で、LOB列の値はNULLになります。

• タスクが予期せず中断され、再開点からの転送を続行する場合、RocketMQインスタンスには一部の重複データ（直近1分以内のもの）が存在する可能性があります。そのため、ダウンストリームシステムは重複除去機能を備えている必要があります。

• データ移行タスクを新規作成する際に、増分同期 のみを設定した場合、データ移行ではソースデータベースのローカル増分ログを48時間以上保持する必要があります。

  データ移行タスクを新規作成する際に、完全移行 + 増分同期 を設定した場合、データ移行ではソースデータベースのローカル増分ログを少なくとも7日間以上保持する必要があります。そうでない場合、増分ログを取得できないためデータ移行タスクが失敗し、ソースとターゲットのデータが一致しなくなる可能性があります。

• OceanBaseデータベースをソースとして増分同期を行う場合、データベーステーブルオブジェクトにUDT列が含まれていると、UDT列をサポートしていないため増分同期タスクが異常終了する可能性があります。

サポートされているソースおよびターゲットインスタンスタイプ

以下の表では、OceanBase MySQL CompatibleおよびOceanBase Oracle Compatibleがサポートするインスタンスタイプには、クラスタインスタンス（トランザクション） と セルフマネージド型データベース が含まれます。

手順

1. データ移行タスクを新規作成します。

   1. OceanBase Cloudコンソールにログインします。

   2. 左側のナビゲーションバーで、データサービス > データ移行 をクリックします。

   3. データ移行 ページで、データ移行 タブをクリックします。

   4. データ移行 タブで、右上の タスクを作成する をクリックします。

2. タスク名を編集するテキストボックスに、カスタムの移行タスク名を入力します。

   数字と英字の組み合わせを使用することを推奨します。名前にはスペースを含めず、長さは64文字を超えないようにしてください。

3. ソースとターゲットを設定する ページで、各パラメータを設定します。

4. ソース エリアで、各パラメータを設定します。

   既に作成・保存されたデータソースを参照する必要がある場合、ソース の右側にある クイック入力 をクリックし、ドロップダウンリストから対象のデータソースを選択します。選択完了後、ソースエリアの各設定は自動的に入力されます。現在の設定を新しいデータソースとして保存する場合は、ソース エリアの右上にある 保存 アイコンをクリックしてください。

   また、クイック入力 のドロップダウンリストで データソースの管理 をクリックすると、データソース ページに移動し、データソースの表示と管理ができます。このページでは、異なる種類のデータソースを一元的に管理できます。詳細については、データソースモジュールのドキュメントを参照してください。

   パラメータ	説明
   クラウドプロバイダー	現在、AWS と Alibaba Cloud がサポートされています。
   リージョン	ソースデータベースが存在するリージョンを選択します。
   データベースタイプ	実際の状況に応じて、ソースデータベースのタイプを OceanBase MySQL互換モード または OceanBase Oracle互換モード に設定します。
   インスタンスタイプ	クラスタインスタンス（トランザクション） または セルフマネージド型データベース を選択します。
   インスタンス	OceanBaseデータベースが存在するインスタンスのIDまたは名前です。クラスタリスト ページで、対象インスタンスのIDまたは名前を確認できます。 説明 クラウドプロバイダーがAlibaba Cloudの場合、アカウント間の権限付与が可能なAlibaba Cloudのメインアカウントインスタンスの選択もサポートされています。詳細については、Alibaba Cloudアカウントの権限付与を参照してください。
   テナント	OceanBaseデータベースのテナントIDまたは名前です。インスタンス ページで、対象インスタンスを展開し、そのインスタンス下の対象テナントのIDまたは名前を確認できます。
   アカウント	データ移行用のOceanBaseデータベースのユーザー名です。
   パスワード	データベースユーザーのパスワードです。

   ソースが セルフマネージド型データベース インスタンスタイプの場合、構造移行と増分同期を行う必要がある場合は、詳細設定 エリアで各パラメータを設定してください。

   • スキーマ移行 ページで 増分同期 または タイプとオブジェクトを選択 を選択する必要がある場合は、sysテナントアカウントを有効にし、以下のパラメータを設定してください。

     パラメータ	説明
     sysアカウント	sysユーザーの名前です。このユーザーは主にOceanBaseデータベースの増分ログとデータベースオブジェクト構造情報の読み取りに使用されます。業務クラスタのsysテナント内に作成してください。
     パスワード	sysユーザーのパスワードです。

   • 増分同期 ページで タイプとオブジェクトを選択 を選択する必要がある場合は、増分ログプロキシサービスを有効にし、OBLogProxy接続情報 を入力してください。

     注意

     sysテナントアカウントと増分ログプロキシサービスの両方を有効にする必要があります。これにより、増分同期がサポートされます。

     OBLogProxy 接続情報 は、OceanBaseデータベースの増分ログプロキシサービスであり、サービス形態でリアルタイムの増分プロジェクトへのアクセスと管理機能を提供し、アプリケーションがOceanBaseデータベースの増分ログにアクセスすることを容易にします。同時に、ネットワークが分離されている状況でも、増分ログのサブスクリプションニーズを解決できます。形式は OBLogProxy IP:OBLogProxy Port です。

5. ターゲット エリアで、各パラメータを設定します。

   既に作成して保存したデータソースを参照する必要がある場合は、ターゲット の右側にある クイック入力 をクリックし、ドロップダウンリストから対象のデータソースを選択します。選択完了後、ターゲット側エリアの各設定は自動的に入力されます。現在の設定を新しいデータソースとして保存する場合は、ターゲット エリアの右上にある 保存 アイコンをクリックします。

   また、クイック入力 のドロップダウンリストで データソースの管理 をクリックすると、データソース ページに移動し、データソースの表示と管理を行えます。このページでは、異なるタイプのデータソースを統一的に管理できます。詳細については、データソース モジュールのドキュメントを参照してください。

   パラメータ	説明
   クラウドプロバイダー	現在、AWS と Alibaba Cloud をサポートしています。
   リージョン	ターゲットインスタンスのリージョンを選択します。
   データベースタイプ	ターゲットを RocketMQ に設定します。
   インスタンスタイプ	選択したクラウドプロバイダーのサポート状況に応じて、クラウドインスタンス または セルフマネージド型RocketMQ を選択します。
   接続タイプ	プライベートネットワーク接続 と パブリック接続 が含まれます。 プライベートネットワーク接続 を選択した場合、ページに表示されているアカウントIDをエンドポイントサービスの許可リストに追加する必要があります。これにより、そのアカウントのエンドポイントがエンドポイントサービスに接続できるようになります。詳細については、プライベートネットワーク接続の選択 モジュールのドキュメントを参照してください。 AWS を クラウドプロバイダー として選択し、エンドポイントサービスの作成時に、エンドポイント接続の承諾を要求 パラメータで 必要 を選択した場合、データ移行サービスが初めてプライベート接続にアクセスする際、AWSコンソールにアクセスし、受信した接続リクエストに対して エンドポイント接続リクエストの受け入れ 操作を行うよう求められます。 パブリックネットワーク接続 を選択した場合、ページに表示されているデータソースIPアドレスをRocketMQインスタンスの許可リストに追加して接続性を確保する必要があります。詳細については、パブリックネットワーク接続の選択 モジュールのドキュメントを参照してください。 説明 ソース側とターゲット側のリージョンを選択すると、ページに許可リストに追加する必要があるデータソースIPアドレスが表示されます。
   接続情報	接続タイプ を プライベートネットワーク接続 に設定した場合、namesrv設定とbroker設定を入力してください。これには、エンドポイントサービスノードとRocketMQアクセスポイントが含まれます。 接続タイプ を パブリックネットワーク接続 に設定した場合、RocketMQアクセスポイントを入力してください。 詳細については、データソースの新規作成 のRocketMQデータソースの章を参照してください。

   RocketMQデータソースのセキュリティ認証設定が完了していない場合、そのデータソースをソースまたはターゲットとするデータ移行タスクが接続失敗する可能性があります。ACLセキュリティ認証を有効にして設定することができます。

   パラメータ	説明
   AccessKey	クラウドコンソールで作成されたAccessKey IDで、認証に使用されます。
   SecretKey	クラウドコンソールで作成されたAccessKey Secretで、認証に使用されます。

6. テストして続行 をクリックします。

   タイプとオブジェクトを選択 ページで、各パラメータを設定します。

   1. 移行タイプ セクションで、現在のデータ移行タスクの移行タイプを選択します。

      パラメータ	説明
      フル移行	フル移行タスク開始後、データ移行サービスはソースデータベースのテーブルの既存データをターゲットデータベースの対応するTopicに移行します。
      増分同期	増分同期タスク開始後、データ移行はソースデータベースで変更されたデータ（追加、変更、削除）をターゲットデータベースの対応するTopicに同期します。現在は DML同期 のみをサポートしており、ニーズに応じてカスタマイズ設定できます。詳細については、DML/DDLのカスタム設定を参照してください。

   2. 移行対象を選択してください セクションで、移行オブジェクトの選択方法を設定します。

      オブジェクトの指定 と マッチングルール の2つの方法で移行オブジェクトを選択できます。

   3. 移行範囲を選択してください セクションで、移行するオブジェクトを選択します。

      • オブジェクトの指定 を選択した場合、データ移行は データベースオブジェクト と データベース全体 をサポートしています。データベーステーブル移行では、1つまたは複数のデータベースのテーブルを移行オブジェクトとして選択できます。データベース移行では、データベース全体を移行オブジェクトとして選択できます。あるデータベースでデータベーステーブル移行を選択した場合、そのデータベースはデータベース移行をサポートしなくなります。あるデータベースでデータベース移行を選択した場合、そのデータベースはデータベーステーブル移行をサポートしなくなります。

        OceanBaseデータベースのデータをRocketMQに移行する際、複数のテーブルから複数のTopicへの移行をサポートしています。

        1. 選択エリアの左側で、移行するオブジェクトを選択します。

        2. > をクリックします。

        3. トピック名 のポップアップウィンドウ内の オブジェクトをトピックにマッピングする ダイアログで、手動入力または検索して指定したTopicを選択します。

        4. OK をクリックします。

        データ移行は、テキストを通じてオブジェクトのインポートをサポートし、ターゲット側のオブジェクトに対してTopicの変更、行フィルタリングの設定、単一オブジェクトまたはすべてのオブジェクトの削除などの操作をサポートします。ターゲット側のオブジェクトの構造はTopic > Database > Tableです。

        操作	説明
        オブジェクトのインポート	選択エリアの右側のリストで、右上の インポート対象 をクリックします。詳細については、移行オブジェクトのインポートを参照してください。
        Topicの変更	データ移行では、ターゲットオブジェクトのTopicを変更する操作をサポートしています。詳細については、Topicの変更を参照してください。
        設定	データ移行では、WHERE 条件を使用した行フィルタリング、シャーディング列と同期が必要な列の選択をサポートしています。**設定**ダイアログで、以下の操作を実行できます。 行フィルターエリアのテキストボックスに WHERE 条件を入力して行フィルタリングを実行します。詳細については、SQL条件によるデータのフィルタリングを参照してください。 シャーディング列ドロップダウンリストから、ターゲットのシャーディング列を選択します。複数のフィールドをシャーディング列として選択できます。このパラメータはオプションです。 シャーディング列を選択する際、特別な場合がない限り、主キーを選択するのがデフォルトです。主キーの負荷が不均衡な場合は、一意の識別子で負荷が比較的均等なフィールドをシャーディング列として選択し、潜在的なパフォーマンス問題を回避してください。シャーディング列の主な役割は以下の通りです： 負荷分散：ターゲット側で同時書き込みが可能な場合、シャーディング列によってメッセージ送信に使用する特定のスレッドを区別します。 順序性：同時書き込みによって生じる可能性のある無秩序な問題を防ぐため、データ移行はシャーディング列の値が同じ場合、ユーザーが受信するメッセージが順序付けられていることを保証します。ここでの順序とは、変更の順序（DMLが列に対して実行される順序）を指します。 列の選択エリアで、同期が必要な列を選択します。詳細については、列のフィルタリングを参照してください。
        削除／すべてクリア	データ移行では、データマッピング時に、一時的にターゲット側に選択された単一または複数のオブジェクトを削除する操作をサポートしています。 単一移行オブジェクトの削除 選択エリアの右側のリストで、ターゲットオブジェクトの後ろにある 削除 アイコンをクリックすると、その移行オブジェクトを削除できます。 すべての移行オブジェクトの削除 選択エリアの右側のリストで、右上の すべて削除 をクリックします。ダイアログで OK をクリックすると、すべての移行オブジェクトが削除されます。

      • **ルールによる一致**を選択します。詳細については、データベースからメッセージキューへのマッチングルールの設定を参照してください。

7. **次へをクリックします。移行オプション**ページで、各パラメータを設定します。

   • フル移行

     **完全移行手順で、タイプとオブジェクトを選択**を選択した場合にのみ、以下のパラメータが表示されます。

     パラメータ	説明
     読み取り並列設定	このパラメータは、フル移行段階でソース側からデータを読み取る並列数を設定するために使用されます。最大制限は512です。並列数が高すぎると、ソース側の負荷が過大になり、業務に影響を与える可能性があります。
     書き込み並列設定	このパラメータは、フル移行段階でターゲット側にデータを書き込む並列数を設定するために使用されます。最大制限は512です。並列数が高すぎると、ターゲット側の負荷が過大になり、業務に影響を与える可能性があります。
     フル移行レートの制限	実際のニーズに応じて、フル移行レートの制限を有効にするかどうかを決定できます。有効にする場合は、ソース側の読み取りRPS（フル移行段階で1秒間にソース側から読み取れる行数の最大制限）、ソース側の読み取りBPS（フル移行段階で1秒間にソース側から読み取れる流量の最大制限）、ターゲット側の書き込みRPS（フル移行段階で1秒間にターゲット側に書き込める行数の最大制限）、ターゲット側の書き込みBPS（フル移行段階で1秒間にターゲット側に書き込める流量の最大制限）を設定してください。 説明 ここで設定されたRPSとBPSは、速度制限とトラフィック制限の能力としてのみ機能します。フル移行が実際に達成できる性能は、ソース側、ターゲット側、インスタンス仕様の設定などの要因の影響を受けます。

   • 増分同期

     **増分同期手順で、タイプとオブジェクトを選択**を選択した場合にのみ、以下のパラメータが表示されます。

     パラメータ	説明
     書き込み並列設定	このパラメータは、増分同期段階でターゲット側にデータを書き込む並列数を設定するために使用されます。最大制限は512です。並列数が高すぎると、ターゲット側の負荷が過大になり、業務に影響を与える可能性があります。
     増分移行レートの制限	実際のニーズに応じて、増分同期レートの制限を有効にするかどうかを決定できます。有効にする場合は、ターゲット側の書き込みRPS（増分同期段階で1秒間にターゲット側に同期できるデータ行数の最大値制限）とターゲット側の書き込みBPS（増分同期段階で1秒間にターゲット側に同期できる流量の最大値制限）を設定してください。 説明 ここで設定されたRPSとBPSは、速度制限とトラフィック制限の能力としてのみ機能します。増分同期が実際に達成できる性能は、ソース側、ターゲット側、インスタンス仕様の設定などの要因の影響を受けます。
     増分同期開始時刻	移行タイプの選択時に フル移行 を選択した場合、このパラメータは表示されません。 移行タイプの選択時に フル移行 を選択せず、増分同期 を選択した場合は、特定の時点以降のデータの移行を指定してください。デフォルトは現在のシステム時刻です。詳細については、増分同期時刻の設定を参照してください。

   • 詳細オプション

     パラメータ	説明
     シリアライズ方式	データ移行時のRocketMQへのメッセージ形式を制御します。現在、Default、Canal、DataWorks(V2.0をサポート)、SharePlex、DefaultExtendColumnType、Debezium、DebeziumFlatten、DebeziumSmtをサポートしています。詳細については、データ形式の説明を参照してください。 説明 現在、OceanBaseデータベースのMySQL互換モードのみが Debezium、DebeziumFlatten、DebeziumSmt をサポートしています。 DataWorks を選択した場合、DDLの同期は COMMENT ON TABLE と ALTER TABLE…TRUNCATE PARTITION をサポートしません。
     パーティションルール	ソース側のデータをRocketMQに同期するルールです。現在、Hashのみをサポートしています。Hashは、一定のHashアルゴリズムを使用してデータ転送を行い、主キー値またはシャーディング列値に基づいてHashを計算し、RocketMQのキュー(MessageQueue)を選択します。
     業務システム識別子(オプション)	**シリアル化方法**がDataWorksに設定されている場合にのみ、このパラメータが表示されます。データの業務システムの出所を識別するために使用され、後のカスタム処理に役立ちます。この業務システム識別子の長さは1～20文字に制限されます。

8. **次へ**をクリックすると、システムはデータ移行タスクの事前チェックを実行します。

   **事前チェック**の段階で、データ移行はターゲット側のRocketMQとの接続状況を検出します。事前チェックでエラーが発生した場合：

   • 問題を調査・処理した後、事前チェックを再実行し、成功するまで繰り返すことができます。

   • 失敗した事前チェック項目の操作列にある スキップ をクリックすることもできます。これにより、その操作をスキップする具体的な影響を示すダイアログがポップアップ表示されます。スキップしてもよいことを確認したら、ダイアログの OK をクリックしてください。

9. 事前チェックが成功したら、**購入をクリックし、データ移行インスタンスを購入する**ページで購入を行います。

   購入が完了すると、データ移行タスクを開始できます。購入の詳細については、データ移行インスタンスの購入を参照してください。現時点でデータ移行インスタンスを購入する必要がない場合は、保存 をクリックし、データ移行タスクの詳細ページに移動してください。後から必要に応じて手動で購入できます。

   データ移行サービスでは、データ移行タスクの実行中に移行オブジェクトを変更できます。詳細については、移行オブジェクトの表示と変更を参照してください。データ移行タスクが開始されると、選択した移行タイプに従って順次実行されます。詳細については、データ移行タスクの詳細を表示するの「移行の詳細を表示する」モジュールの内容を参照してください。