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

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

背景情報

Kafkaは現在広く利用されている高性能な分散ストリーム処理プラットフォームです。データ移行サービスは、OceanBaseデータベースの2つの互換モードとKafkaデータソース間のリアルタイムデータ同期をサポートし、メッセージ処理能力を拡張します。リアルタイムデータウェアハウスの構築、データクエリ、レポート処理の負荷分散などのビジネスシナリオで広く利用されています。

Kafkaネットワーク接続タスクには、クラウドインスタンスKafkaネットワークとセルフマネージドKafkaネットワークが含まれます。異なるタイプのネットワーク接続タスクのプロセスは以下のとおりです。

• クラウドインスタンスKafkaネットワーク接続タスク

• セルフマネージドKafkaネットワーク接続タスク

前提条件

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

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

制限事項

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

• ソース側データベースの操作制限

  全量移行段階では、データベースやテーブル構造を変更するDDL操作を実行しないでください。実行した場合、データ移行タスクが中断する可能性があります。

• 現在、データ移行はKafka V0.9、V1.0、V2.x、OceanBaseデータベース（MySQLおよびOracle互換モードを含む）V2.x、V3.x、V4.xをサポートしています。

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

• データ移行でサポートされるオブジェクトには、物理テーブルとマテリアライズドビュー(Materialized View)が含まれます。マテリアライズドビューの同期機能は、現在OceanBase MySQL互換モードからKafkaへの移行経路でのみサポートされており、ソース側のOceanBase MySQL互換モードのバージョンはV4.3.5.6～V4.4.0、またはV4.4.2.4以上である必要があります。

• データ移行中に、ソース側で移行対象範囲内のテーブル名を変更し、その後の名前が移行オブジェクトに含まれない場合、そのデータはターゲットKafkaインスタンスに同期されません。

• マテリアライズドビュー同期の制限事項は以下のとおりです。

  • データベース全体の移行およびマッチングルールを使用する場合、タスク実行中に新規作成されたマテリアライズドビューの増分DMLは自動的にKafkaに配信されますが、そのマテリアライズドビューを作成したDDLは配信されません。

  • マテリアライズドビュー同期は、マテリアライズドビューのDMLデータの同期のみをサポートし、マテリアライズドビューを作成したDDLの同期は行いません。

  • COMPLETE方式でマテリアライズドビューをリフレッシュした後、リフレッシュ結果の変更をKafkaに同期することはサポートされていません。マテリアライズドビューでREFRESH FORCEを使用し、実際のリフレッシュ方式がCOMPLETEにフォールバックした場合も同様の制限が適用されます。

• マテリアライズドビュー同期機能を使用する前に、ソース側のOceanBaseクラスタバージョンがV4.3.5.6～V4.4.0、またはV4.4.2.4以上であることを確認してください。既存のクラスタの場合、マテリアライズドビュー同期をサポートするクラスタバージョンにアップグレードした後、OMSを使用してマテリアライズドビューを同期する前に、アップグレード後のクラスタバージョンに応じた操作を完了する必要があります。

  • アップグレード後のクラスタバージョンがV4.3.5.6～V4.4.0の場合：

    1. チケットを提出し、OMSマテリアライズドビュー同期機能の有効化を申請します。

    2. データディクショナリの定期DUMPの実行間隔を1分に調整します：

       CALL DBMS_DATA_DICT.MODIFY_DUMP_INTERVAL(INTERVAL_EXPR);
       

    3. DUMP実行後、以下のSQLで新しいSCNが生成されたか確認します：

       SELECT * FROM CDB_OB_DATA_DICTIONARY_IN_LOG LIMIT 5;
       

    4. 新しいSCNが生成されたことを確認したら、データディクショナリの定期DUMPの実行間隔を適切な値に調整します。例えば2時間などです。

    5. 上記の操作が完了したら、OMSタスクを作成または設定してマテリアライズドビュー同期を開始します。

  • アップグレード後のクラスタバージョンがV4.4.2.4以上の場合：

    1. OMSタスクの作成または設定前に、データディクショナリDUMPを手動で1回トリガーします：

       CALL DBMS_DATA_DICT.TRIGGER_DUMP;
       

    2. DUMP完了後、OMSタスクを作成または設定してマテリアライズドビュー同期を開始します。

注意事項

• ノード間のクロックが同期していない場合、またはコンピュータ端末とサーバー間のクロックが同期していない場合、増分同期の遅延時間が正確でない可能性があります。

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

• ソース側が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になります。

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

• OceanBaseデータベースのデータをKafkaに同期する際、ソース側で一意インデックス作成文の実行に失敗した場合、Kafkaは作成DDL文と削除DDL文を消費します。下流に伝送されたインデックス作成DDLの実行に失敗した場合は、この例外を無視してください。

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

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

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

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

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

操作手順

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

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

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

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

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

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

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

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

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

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

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

   パラメータ	説明
   クラウドベンダー	現在、Alibaba Cloud、AWS、および Google Cloud のみサポートしています。
   データベースタイプ	データベースタイプを選択してください。ソース側を OceanBase MySQL互換モード または OceanBase Oracle互換モード に設定します。
   インスタンスタイプ	インスタンスタイプを選択してください。現在、専用インスタンス(トランザクション型) と セルフマネージドデータベース をサポートしています。
   リージョン	リージョンを選択してください。ソース側データベースが存在するリージョンを選択します。
   接続タイプ	接続タイプを選択してください。エンドポイント と パブリック接続 が含まれます。 エンドポイント 接続方式を選択した場合、ページに表示されているアカウントIDを、エンドポイントサービスの許可リストに追加する必要があります。これにより、そのアカウントのエンドポイントがエンドポイントサービスに接続できるようになります。詳細については、プライベートネットワーク接続の選択モジュールのドキュメントを参照してください。 パブリックネットワーク接続 接続方式を選択した場合、ページに表示されているデータソースIPアドレスをOceanBaseデータベースインスタンスの許可リストに追加する必要があります。これにより、接続性が確保されます。詳細については、パブリックネットワーク接続の選択モジュールのドキュメントを参照してください。 説明 このパラメータは、インスタンスタイプがセルフマネージドデータベースの場合にのみ表示されます。ソース側とターゲット側のリージョンを選択すると、ページに許可リストに追加する必要があるデータソースIPアドレスが表示されます。
   接続情報	接続タイプを選択してください。インスタンスタイプがセルフマネージドデータベースの場合にのみこのパラメータが表示されます。 接続タイプ で エンドポイント を選択した場合、エンドポイントサービス名を入力してください。 接続タイプ で パブリックネットワーク接続 を選択した場合、データベースホストのIPアドレスとポート番号を入力してください。
   インスタンス	インスタンスを選択してください。OceanBaseデータベースが存在するインスタンスのIDまたは名前です。クラスターリスト ページで、ターゲットインスタンスのIDまたは名前を確認できます。 説明 クラウドベンダーがAlibaba Cloudの場合、アカウント間のアクセス許可が設定されたAlibaba Cloudのメインアカウントのインスタンスの選択も可能です。詳細については、Alibaba Cloudアカウントのアクセス許可を参照してください。
   テナント	テナントを選択してください。OceanBaseデータベーステナントのIDまたは名前です。インスタンス ページで、ターゲットインスタンスを展開し、そのインスタンス下のターゲットテナントのIDまたは名前を確認できます。
   データベースアカウント	データ移行用のOceanBaseデータベースのユーザー名。
   パスワード	データベースユーザーのパスワード。

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

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

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

   • **タイプとオブジェクトを選択で増分同期**を選択する場合は、OBLogProxyを有効にし、OBLogProxy接続情報を入力してください。

     説明

     増分同期をサポートするには、sysテナントアカウントとOBLogProxyの両方を有効にする必要があります。

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

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

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

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

   パラメータ	説明
   クラウドベンダー	現在、Alibaba Cloud、AWS、および Google Cloud のみをサポートしています。ソース側と同じクラウドベンダーを選択することも、異なるクラウドベンダー間のデータ移行も可能です。 説明 現在、異なるクラウドベンダー間のデータ移行機能はデフォルトで利用できません。この機能を使用する場合は、OceanBase Cloudのテクニカルサポートにお問い合わせください。
   データベースタイプ	ターゲット側に Kafka を選択します。
   インスタンスタイプ	クラウドインスタンス と セルフマネージド Kafka をサポートしています。
   リージョン	ターゲット側インスタンスのリージョンを選択します。
   接続タイプ	エンドポイント と パブリックネットワーク接続 が含まれます。クラウドベンダー を Alibaba Cloud、AWS、または Google Cloud に設定した場合、ピアリング接続 もサポートされます。 エンドポイント 接続方式を選択した場合、ページに表示されているアカウントIDを、エンドポイントサービスの許可リストに追加する必要があります。これにより、そのアカウントのエンドポイントがエンドポイントサービスに接続できるようになります。詳細については、プライベートネットワーク接続の選択モジュールのドキュメントを参照してください。 AWS を クラウドベンダー として選択し、エンドポイントサービスの作成時に、エンドポイント接続の承諾を要求 パラメータで 必要 を選択した場合、データ移行サービスが初めてプライベート接続にアクセスする際、AWSコンソールにアクセスし、受信したConnectionsリクエストに対して エンドポイント接続リクエストの受け入れ 操作を行うよう求められます。 クラウドベンダーを Google Cloud として選択した場合、Published Servicesに認可プロジェクトを追加してください。認可後は、データソースの接続テスト時に手動で認可する必要はありません。 パブリックネットワーク接続 接続方式を選択した場合、ページに表示されているデータソースIPアドレスをKafkaインスタンスの許可リストに追加して、接続性を確保する必要があります。詳細については、パブリックネットワーク接続の選択モジュールのドキュメントを参照してください。 説明 ソース側とターゲット側のリージョンを選択すると、ページに許可リストに追加する必要があるデータソースIPアドレスが表示されます。 ピアリング接続 を選択した場合、ドロップダウンリストから利用可能なピアリング接続を選択し、ページに表示されているIPアドレスをKafkaデータベースインスタンスのセキュリティグループのインバウンドルールまたは許可リストに追加して、接続性を確保してください。具体的な操作については、AWSピアリング接続、Alibaba Cloudピアリング接続、および Google Cloudピアリング接続を参照してください。
   ピアリング接続ID	接続タイプが ピアリング接続 の場合にのみ表示されます。ドロップダウンリストから利用可能なピアリング接続を選択します。ドロップダウンリストには、現在のクラウドベンダーとリージョンが一致し、ステータスが 接続済み のピアリング接続のみが表示され、対応するVPC IDとCIDRが表示されます。利用可能な接続がない場合、新規 をクリックして、新しいタブでデータサービスのピアリング接続を作成できます。
   接続情報	接続タイプ で エンドポイント を選択した場合、エンドポイントサービス名とKafkaアクセスポイントを入力してください。 接続タイプ で パブリックネットワーク接続 を選択した場合、Kafkaアクセスポイントを入力してください。 接続タイプ で ピアリング接続 を選択した場合、Kafkaアクセスポイントを入力してください。

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

   1. ファイルをアップロード をクリックし、拡張子が .jks の信頼済み証明書をアップロードします。

   2. 業務ニーズに応じて、SASL を有効にするかどうかを選択します。SASLを有効にする場合は、以下のパラメータを設定してください。

      パラメータ	説明
      認証方式	PLAIN、SCRAM-SHA-256、および SCRAM-SHA-512 をサポートします。 PLAIN認証方式は比較的シンプルですが、ユーザーを動的に変更できず、ユーザー名とパスワードを平文で設定するため、セキュリティは高くありません。 SCRAM（Salted Challenge Response Authentication Mechanism）認証方式は、ユーザー名とパスワードによる認証の従来の仕組みを実行することで、セキュリティ上の問題を解決します。KafkaはSCRAM-SHA-256をサポートしており、TLSと組み合わせてセキュアな認証を実行できます。 この認証方式ではユーザーを動的に変更でき、ユーザーデータはZookeeperに保存されます。Brokerを起動する前に、まずZookeeperと通信し、Brokerとの通信用ユーザーを作成する必要があります。ただし、この認証方式ではユーザー名とパスワードを平文で設定する必要があります。 KafkaはSCRAM-SHA-512をサポートしており、TLSと組み合わせてセキュアな認証を実行できます。
      ユーザー名	データ移行に使用するユーザー名。
      パスワード	データ移行に使用するユーザーのパスワード。

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

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

   1. 移行タイプ エリアで、現在のデータ移行タスクの移行タイプを選択します。

      **移行タイプには、全量移行と増分同期**が含まれます。

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

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

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

   3. 移行範囲を選択してください エリアで、移行するオブジェクトを選択します。

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

        OceanBaseデータベースのデータをKafkaに移行する場合、複数のテーブルから複数のTopicへの移行がサポートされます。

        オブジェクト選択パネルには、**Tableタイプの通常のテーブルが表示されます。ソース側のバージョンがマテリアライズドビュー機能をサポートする場合、Materialized view**が表示され、展開すると現在のデータベースのマテリアライズドビューを確認および選択できます。

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

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

        3. オブジェクトをトピックにマッピングする ダイアログの 既存のトピック ドロップダウンリストで、移行するTopicを検索して選択します。

        4. 確認 をクリックします。

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

        説明

        データベース全体の移行 を選択した場合：

        • 右側のリストにはデータベース名のみが表示され、具体的なオブジェクトは表示されません。

        • 増分同期-DDL同期 の移行タイプを選択した場合、ソース側で新規作成されたテーブルはターゲット側に同期できます。

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

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

        ソース側でマテリアライズドビュー機能の利用条件を満たしている場合、マッチングルールはテーブルとマテリアライズドビューの両方を同時にマッチングできます。新しいマッチングルールを作成する際、デフォルトで テーブル と マテリアライズドビュー の2種類の移行オブジェクトタイプが選択されていますが、どちらか一方のみを選択することもできます。マッチング結果はタイプごとに分類表示されます。

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

   • 全量移行

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

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

   • 増分同期

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

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

   • 詳細オプション

     パラメータ	説明
     シリアライズ方式	Kafkaへのデータ移行時のメッセージ形式を制御します。現在、Default、Canal、DataWorks(V2.0をサポート)、SharePlex、DefaultExtendColumnType、Debezium、DebeziumFlatten、DebeziumSmt、Avroをサポートしています。データ形式の説明を参照してください。 説明 現在、Debezium、DebeziumFlatten、DebeziumSmt、Avro は、OceanBaseデータベースのMySQL互換モードでのみサポートされています。 DataWorks を選択した場合、DDLの同期は COMMENT ON TABLE と ALTER TABLE…TRUNCATE PARTITION をサポートしません。
     パーティションルール	OceanBaseデータベースのデータをKafka Topicに同期するルールです。現在、Hash、Table、Oneをサポートしています。異なるシナリオでのDDLステートメントの配信と例については、表の下部の説明を参照してください。 Hashは、データ移行が一定のHashアルゴリズムを使用し、主キー値またはシャーディング列値のHashに基づいてKafka Topicのパーティションを選択することを示します。 Tableは、データ移行がテーブル内のすべてのデータを同じパーティションに配信し、テーブル名をHashキーとすることを示します。 Oneは、JSONメッセージがTopicの下の特定のパーティションに配信されることを示します。これは、メッセージの順序を維持するためです。
     業務システム識別子(オプション)	**シリアル化方法**でDataWorksを選択した場合にのみ、このパラメータが表示されます。データの生成元である業務システムを識別するために使用され、後のカスタム処理に役立ちます。この業務システム識別子の長さは1〜20文字に制限されます。

     以下の表は、異なるシナリオにおけるDDLステートメントの配信方法を示しています。

     パーティションルール	DDLステートメントが複数のテーブルに関連する場合（例：RENAME TABLE）	DDLステートメントで関連テーブルを特定できない場合（例：DROP INDEX）	DDLステートメントが単一のテーブルに関連する場合
     Hash	DDLステートメントは、関連テーブルが存在するTopicのすべてのパーティションに配信されます。 例えば、DDLステートメントがA、B、Cの3つのテーブルに関連している場合、AがTopic 1に、BがTopic 2に、Cがこのタスクに含まれていない場合、そのDDLステートメントはTopic 1とTopic 2のすべてのパーティションに配信されます。	DDLステートメントは、このタスクのすべてのTopicのすべてのパーティションに配信されます。 例えば、DDLステートメントがデータ移行によって認識されない場合、このタスクに3つのTopicが存在すると、そのDDLステートメントはこれら3つのTopicのすべてのパーティションに配信されます。	DDLステートメントは、そのテーブルが属するTopicのすべてのパーティションに配信されます。
     Table	DDLステートメントは、関連テーブルが存在するTopicの、対応するテーブル名のHash値が存在するパーティションに配信されます。 例えば、DDLステートメントがA、B、Cの3つのテーブルに関連している場合、AがTopic 1に、BがTopic 2に、Cがこのタスクに含まれていない場合、そのDDLステートメントはTopic 1とTopic 2の関連テーブルのHash値が存在するパーティションに配信されます。	DDLステートメントは、このタスクのすべてのTopicのすべてのパーティションに配信されます。 例えば、DDLステートメントがデータ移行によって認識されない場合、このタスクに3つのTopicが存在すると、そのDDLステートメントはこれら3つのTopicのすべてのパーティションに配信されます。	Table Nameに基づいてHashを計算し、そのテーブルが属するTopic内の特定のパーティションに配信されます。
     One	DDLステートメントは、関連テーブルが存在するTopicの固定パーティションに配信されます。 例えば、DDLステートメントがA、B、Cの3つのテーブルに関連している場合、AがTopic 1に、BがTopic 2に、Cがこのタスクに含まれていない場合、そのDDLステートメントはTopic 1とTopic 2の特定の固定パーティションに配信されます。	DDLステートメントは、このタスクのすべてのTopicの特定の固定パーティションに配信されます。 例えば、DDLステートメントがデータ移行によって認識されない場合、このタスクに3つのTopicが存在すると、そのDDLステートメントはこれら3つのTopicの特定の固定パーティションに配信されます。	DDLステートメントは、そのテーブルが属するTopicの特定の固定パーティションに配信されます。

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

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

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

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

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

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

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