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

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

背景

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

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

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

  

• 自社構築Kafkaネットワーク接続タスク

  

前提条件

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

• ソース側のOceanBaseデータベースのMySQL互換モードとターゲット側のKafkaインスタンスに、データ移行専用のデータベースユーザーを作成し、関連する権限を付与していること。詳細については、データ移行ユーザー権限をご参照ください。

使用上の制限

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

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

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

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

• データ移行サービスはクラスタインスタンスのみをサポートしており、テナントインスタンスは現在サポートしていません。

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

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

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

注意事項

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

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

• ソースが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. ソースとターゲットを設定する ページで、各パラメータを設定します。

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

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

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

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

   クイックフィル のドロップダウンリストで、データソースの管理 をクリックすることもできます。これにより、データソース ページに移動し、データソースの表示や管理を行えます。この記事では異なるタイプのデータソースを一元的に管理します。詳細については、データソースモジュールのドキュメントをご参照ください。

   パラメータ	説明
   クラウドベンダー	現在、AWS、Huawei Cloud、Alibaba Cloud、および Tencent Cloud をサポートしています。
   データベースタイプ	ソースを OceanBase MySQL Compatible または OceanBase Oracle Compatible に選択します。
   インスタンスタイプ	現在、クラスタインスタンス（トランザクション） と 自己管理型データベース をサポートしています。
   リージョン	ソースデータベースのリージョンを選択します。
   インスタンス	OceanBaseデータベースが存在するインスタンスのIDまたは名前です。クラスタリスト ページで、対象インスタンスのIDまたは名前を確認できます。 説明 クラウドベンダーがAlibaba Cloudの場合、アカウント間認証をサポートするAlibaba Cloudのメインアカウントインスタンスの選択も可能です。詳細については、Alibaba Cloudアカウント認証をご参照ください。
   テナント	OceanBaseデータベースのテナントIDまたは名前です。インスタンス ページで、対象インスタンスを展開し、そのインスタンス下の対象テナントのIDまたは名前を確認できます。
   データベースアカウント	データ移行用のOceanBaseデータベースのユーザー名。
   パスワード	データベースユーザーのパスワード。

   ソースが オンプレデータベース インスタンスタイプの場合、構造移行と増分同期を実行する必要がある場合は、詳細設定 エリアで各パラメータを設定してください。

   

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

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

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

     注意

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

     OBLogProxy 接続情報 は、OceanBaseデータベースの増分ログプロキシサービスであり、サービス形態でリアルタイムの増分ログの取り込みと管理機能を提供し、アプリケーションがOceanBaseデータベースの増分ログにアクセスしやすくします。また、ネットワークが分離されている環境でも、増分ログのサブスクリプションニーズを解決できます。形式は OBLogProxy IP：OBLogProxy Port です。

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

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

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

   パラメータ	説明
   クラウドプロバイダー	現在、AWS、Huawei Cloud、Alibaba Cloud、および Tencent Cloud をサポートしています。ソースと同じクラウドプロバイダーを選択することも、異なるクラウド間でのデータ移行も可能です。 説明 現在、クラウド間のデータ移行機能はデフォルトで無効です。この機能を使用する場合は、OB Cloudテクニカルサポートにお問い合わせください。
   データベースタイプ	ターゲットを Kafka に設定します。
   インスタンスタイプ	クラウドインスタンス Kafka と セルフマネージド Kafka をサポートしています。
   リージョン	ターゲットインスタンスのリージョンを選択します。
   接続タイプ	エンドポイント と パブリックIP が含まれます。 エンドポイント 接続方式を選択した場合、ページに表示されているアカウントIDをエンドポイントサービスのホワイトリストに追加する必要があります。これにより、そのアカウントのエンドポイントがエンドポイントサービスに接続できるようになります。詳細については、プライベートネットワーク接続の選択 モジュールのドキュメントをご参照ください。 クラウドベンダー を AWS に設定した場合、エンドポイントサービス作成時にパラメータ エンドポイントを使用するために受け入れる必要があります で 有効にする を選択した場合、データ移行サービスが初めてプライベート接続にアクセスする際、AWSコンソールにアクセスし、受信した接続リクエストに対して エンドポイント接続リクエストを受け入れる 操作を行うよう求められます。 パブリックIP 接続方式を選択した場合、ページに表示されているデータソースのIPアドレスをKafkaインスタンスのホワイトリストに追加する必要があります。これにより、接続性が確保されます。詳細については、パブリックネットワーク接続の選択 モジュールのドキュメントをご参照ください。 説明 ソースとターゲットのリージョンを選択すると、ページにホワイトリストに追加する必要があるデータソースのIPアドレスが表示されます。
   接続情報	接続タイプ を エンドポイント に設定した場合、エンドポイントサービス名とKafkaアクセスポイントを入力してください。 接続タイプ を パブリックIP に設定した場合、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への移行をサポートします。

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

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

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

        4. わかりました をクリックします。

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

        説明

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

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

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

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

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

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. 事前チェックが成功したら、購入 をクリックし、データ移行インスタンスを購入する ページで購入を行います。

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

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