Data Transmission Service:RDS for MySQL から MaxCompute へのデータ同期

注意事項

• 完全データ初期化中、DTS はソースデータベースとターゲットデータベースの読み取りおよび書き込みリソースを消費し、それらの負荷を増大させます。データベースのパフォーマンスが低い、仕様が低い、またはビジネストラフィックが多い場合 (たとえば、ソースデータベースに多くの低速 SQL クエリやプライマリキーのないテーブルがある場合、またはターゲットデータベースでデッドロックが発生する場合)、データベースの負荷が増加し、サービス中断を引き起こすことさえあります。データを同期する前に、ソースインスタンスとターゲットインスタンスのパフォーマンスを評価してください。両方のインスタンスの CPU 使用率が 30% 未満であるなど、オフピーク時にデータを同期することを推奨します。

• DTS はテーブルレベルでのみデータ同期をサポートします。

• データ同期中に、gh-ost や pt-online-schema-change などのツールを使用して同期オブジェクトに対してオンライン DDL 変更を実行しないでください。実行した場合、同期タスクは失敗します。

• MaxCompute はプライマリキー制約をサポートしていません。ネットワークの問題により DTS がデータ転送をリトライした場合、MaxCompute に重複レコードが出現する可能性があります。

課金

サポートされるソースインスタンスタイプ

DTS は、以下のタイプの MySQL データベースからのデータ同期をサポートしています。

• Elastic Compute Service (ECS) インスタンス上の自己管理データベース

• Express Connect、VPN Gateway、または Smart Access Gateway を介して接続された自己管理データベース

• データベースゲートウェイを介して接続された自己管理データベース

• 同じまたは異なる Alibaba Cloud アカウントに属する ApsaraDB RDS for MySQL インスタンス

サポートされる SQL 操作

• DDL 操作：ALTER TABLE ADD COLUMN

• DML 操作：INSERT、UPDATE、DELETE

同期プロセス

1. 初期スキーマ同期。

   DTS はソーステーブルのスキーマ定義を MaxCompute に同期します。この初期ステップで、DTS はテーブル名に _base サフィックスを追加します。たとえば、customer という名前のソーステーブルは、MaxCompute では customer_base になります。

2. 初期完全データ同期。

   DTS は、ソーステーブルのすべての既存データを、MaxCompute の対応する _base テーブルに同期します。たとえば、customer テーブルのデータは customer_base テーブルに同期されます。これにより、後続の増分データ同期のベースラインが提供されます。

   説明

   このテーブルは、全量ベースラインテーブルとも呼ばれます。

3. 増分データ同期。

   DTS は、MaxCompute に _log サフィックスを持つ増分ログテーブル (例: customer_log) を作成します。その後、ソースからの増分データ変更をこのログテーブルにリアルタイムで同期します。

   説明

   増分ログテーブルのスキーマの詳細については、「増分ログテーブルのスキーマ」をご参照ください。

操作手順

1. データ同期タスクを購入します。詳細については、「購入プロセス」をご参照ください。

   説明

   タスクの購入時に、MySQL パラメーターを MySQL に、MaxCompute パラメーターを MaxCompute に、一方向同期 パラメーターを 一方向同期 に設定します。

2. Data Transmission Service (DTS) コンソールにログインします。

   説明

   Data Management (DMS) コンソールに自動的にリダイレクトされた場合は、右下隅の アイコンをクリックし、次に をクリックして従来の DTS コンソールに戻ることができます。

3. 左側のナビゲーションウィンドウで、データ同期 をクリックします。

4. [同期タスク] ページの上部で、ターゲットインスタンスのリージョンを選択します。

5. 購入したデータ同期インスタンスを見つけ、[タスクの設定] をクリックします。

6. 同期チャネルのソースインスタンスとターゲットインスタンスを設定します。

   

   セクション	パラメーター	説明
   N/A	同期タスク名	DTS は自動的にタスク名を生成します。識別しやすいように、わかりやすい名前を指定することを推奨します。名前は一意である必要はありません。
   ソースインスタンスの詳細	インスタンスタイプ	RDS インスタンス を選択します。
   	インスタンスリージョン	購入時に選択したソースインスタンスのリージョンです。このパラメーターは変更できません。
   	インスタンス ID	ソース RDS インスタンスの ID を選択します。
   	データベースアカウント	ソース RDS インスタンスのデータベースアカウントを入力します。 説明 ソース RDS インスタンスのデータベースタイプが [MySQL 5.5] または [MySQL 5.6] の場合、データベースアカウント および データベースのパスワード パラメーターを設定する必要はありません。
   	データベースパスワード	データベースアカウントのパスワードを入力します。
   	接続モード	要件に応じて、非暗号化 または SSL 暗号化 を選択します。SSL 暗号化 を選択した場合は、事前に RDS インスタンスで SSL 暗号化を有効にする必要があります。詳細については、「SSL 暗号化を設定する」をご参照ください。 重要 暗号化 設定は、中国本土および中国 (香港) リージョンでのみ利用可能です。
   ターゲットインスタンスの詳細	インスタンスタイプ	これは MaxCompute に固定されており、変更できません。
   	インスタンスリージョン	購入時に選択したターゲットインスタンスのリージョンです。このパラメーターは変更できません。
   	プロジェクト	MaxCompute インスタンスの プロジェクト を入力します。これは、MaxCompute コンソールの [プロジェクト] ページで確認できます。

7. ページ右下隅の [ホワイトリストを承認して次へ] をクリックします。

   ソースまたはターゲットデータベースが ApsaraDB RDS for MySQL や ApsaraDB for MongoDB などの Alibaba Cloud データベースインスタンスである場合、DTS は対応するリージョンの DTS サービスの IP アドレスをインスタンスの IP ホワイトリストに自動的に追加します。ソースまたはターゲットデータベースが ECS インスタンス上の自己管理データベースである場合、DTS は対応するリージョンの DTS サービスの IP アドレスを ECS インスタンスのセキュリティグループルールに自動的に追加します。また、データベース設定が ECS インスタンスからのアクセスを許可していることを確認する必要があります。データベースが複数の ECS インスタンスにまたがるクラスターにデプロイされている場合は、他の各 ECS インスタンスのセキュリティグループルールに対応するリージョンの DTS サービスの IP アドレスを手動で追加する必要があります。ソースまたはターゲットデータベースがオンプレミスデータセンター (IDC) 内の自己管理データベース、または他のクラウドプラットフォーム上のデータベースである場合は、DTS サーバーからのアクセスを許可するために、対応するリージョンの DTS サービスの IP アドレスを手動で追加する必要があります。DTS サービスの IP アドレスについては、「DTS サーバーの IP アドレスブロック」をご参照ください。

   警告

   DTS サーバーのパブリック IP アドレス CIDR ブロックを自動または手動で追加すると、セキュリティリスクが生じる可能性があります。このプロダクトを使用することにより、お客様はこれらの潜在的なリスクを認識し、受け入れるものとします。基本的なセキュリティ保護を実装する必要があります。これらの保護には、パスワードセキュリティの強化、CIDR ブロックのオープンポートの制限、内部 API 通信での認証の使用、不要な CIDR ブロックの定期的な確認と制限などが含まれますが、これらに限定されません。あるいは、専用線、VPN Gateway、または Smart Access Gateway などの内部ネットワークを介してデータベースに接続することもできます。

8. ページ右下隅の 次へ をクリックして、MaxCompute プロジェクトに対する以下の権限を DTS 同期アカウントに付与します。

9. 同期ポリシーとオブジェクトを設定します。

   パラメーター	説明
   増分ログテーブルのパーティション定義	ビジネス要件に基づいてパーティション名を選択します。 パーティションの詳細については、「パーティション」をご参照ください。
   同期の初期化	同期の初期化オプションは、初期スキーマ同期と初期完全データ同期です。 [初期スキーマ同期] と [初期完全データ同期] の両方を選択します。DTS は、増分データ同期を開始する前に、選択したオブジェクトのスキーマと既存のデータを同期します。
   既存のターゲットテーブルに対するアクション	エラーの事前チェックと報告：ターゲットデータベースに同じ名前のテーブルが存在するかどうかをチェックします。そのようなテーブルが存在しない場合、チェックは合格します。同じ名前のテーブルが存在する場合、事前チェックは失敗してエラーを報告し、データ同期タスクは開始されません。 説明 ターゲット内の競合するテーブルを簡単に削除または名前変更できない場合は、「オブジェクト名をターゲットインスタンスにマッピングする」ことで名前の競合を回避できます。 エラーを無視して続行：ターゲットデータベースに同じ名前の既存のテーブルがあるかどうかのチェックをスキップします。 警告 エラーを無視して続行 を選択すると、データの不整合やビジネスリスクにつながる可能性があります。例： テーブルスキーマが同一の場合、初期同期フェーズでは、ソースレコードと同じプライマリキー値を持つターゲットのレコードが保持されます。ただし、増分同期フェーズでは、ターゲットのレコードは上書きされます。 テーブルスキーマが異なる場合、データの初期化ができない、一部の列のみが同期される、または同期が完全に失敗する可能性があります。
   同期オブジェクトの選択	ソースオブジェクト ボックスで同期したいテーブルをクリックし、 アイコンをクリックして 選択中のオブジェクト ボックスに移動します。 説明 同期オブジェクトとしてテーブルのみを選択できますが、複数のデータベースからテーブルを選択できます。 デフォルトでは、同期されたオブジェクトの名前は同じままです。ターゲットインスタンスで異なる名前を使用する必要がある場合は、オブジェクト名マッピング機能を使用します。詳細については、「オブジェクト名をターゲットインスタンスにマッピングする」をご参照ください。
   追加列ルールの選択	データを MaxCompute に同期する際、DTS はターゲットテーブルにいくつかの追加列を追加します。これらの追加列がターゲットテーブルの既存の列名と競合する場合、同期は失敗します。ビジネスニーズに基づいて、[追加列の新しい命名規則を有効にする] を ○ または × に設定します。 警告 選択を行う前に、追加列がターゲットテーブルの既存の列名と競合するかどうかを評価してください。名前の競合は、タスクの失敗やデータ損失を引き起こす可能性があります。これらの追加列のルールと定義については、「追加列の名前と定義」をご参照ください。
   マッピング名の変更	ターゲットインスタンスで同期されたオブジェクトの名前を変更するには、オブジェクト名マッピング機能を使用します。詳細については、「データベース、テーブル、および列のマッピング」をご参照ください。
   DMS オンライン DDL 中に一時テーブルをレプリケートする	Data Management Service (DMS) を使用してソースデータベースでオンライン DDL 変更を実行する場合、これらの変更によって生成される一時テーブルを同期するかどうかを指定できます。 ○：オンライン DDL 変更によって生成された一時テーブルからデータを同期します。 説明 オンライン DDL 変更によって一時テーブルに大量のデータが生成されると、データ同期が遅延する可能性があります。 いいえ：一時テーブルからデータを同期しません。ソースデータベースからの元の DDL データのみが同期されます。 説明 このオプションは、ターゲットデータベースでテーブルロックを引き起こします。
   ソース/ターゲットが利用できない場合のリトライ期間	デフォルトでは、DTS がソースまたはターゲットデータベースへの接続に失敗した場合、720 分 (12 時間) リトライします。カスタムのリトライ期間を指定することもできます。指定された期間内に DTS が再接続すると同期タスクは自動的に再開されます。それ以外の場合、タスクは失敗します。 説明 接続リトライ期間中も DTS インスタンスは課金されるため、ビジネス要件に基づいてカスタムのリトライ期間を設定するか、ソースおよびターゲットインスタンスがリリースされた後、速やかに DTS インスタンスをリリースしてください。

10. 上記の設定が完了したら、ページ右下隅の [事前チェックして開始] をクリックします。

    説明

    • DTS は同期タスクが開始される前に事前チェックを実行します。タスクはこのチェックに合格した後にのみ開始できます。

    • 事前チェックが失敗した場合は、失敗した項目の横にある アイコンをクリックして詳細を表示します。

      • 原因に基づいて問題を修正し、再度事前チェックを実行します。

      • アラート項目を修正する必要がない場合は、無視 または [無視して再事前チェック] をクリックしてアラート項目をスキップし、再度事前チェックを実行できます。

11. 事前チェック ダイアログボックスに 事前チェック完了 と表示されたら、事前チェック ダイアログボックスを閉じると、同期タスクが開始されます。

12. タスクが初期化を完了し、同期中 状態になるのを待ちます。

    データ同期 ページでデータ同期タスクのステータスを表示できます。

増分ログテーブルのスキーマ

DTS が MySQL から MaxCompute の増分ログテーブルに増分データを同期する際、次の例に示すように、増分データとメタデータの両方を格納します。

スキーマ定義

before_flag と after_flag フィールド

増分ログの [before_flag] と [after_flag] の値は、操作タイプによって異なります。

• INSERT

  INSERT 操作の場合、すべての列値は新しく挿入されたレコードの値 (「アフター」イメージ) です。したがって、before_flag は N、after_flag は Y です。以下に例を示します。

• UPDATE

  UPDATE 操作の場合、DTS は操作を 2 つのログレコードに分割します。これらの 2 つのレコードは、同じ record_id、operation_flag、および utc_timestamp 値を共有します。

  最初のログレコードは更新前の値 (「ビフォア」イメージ) をキャプチャするため、before_flag は Y、after_flag は N です。2 番目のログレコードは更新後の値 (「アフター」イメージ) をキャプチャするため、before_flag は N、after_flag は Y です。以下に例を示します。

• DELETE

  DELETE 操作の場合、ログレコード内のすべての列値は削除された値 (「ビフォア」イメージ) です。したがって、before_flag は Y、after_flag は N です。以下に例を示します。

完全データと増分データのマージ

データ同期タスクが開始されると、DTS は MaxCompute に全量ベースラインテーブルと増分ログテーブルを作成します。MaxCompute で SQL コマンドを実行してこれら 2 つのテーブルをマージし、特定の時点でのデータの完全なスナップショットを取得できます。

この例では、以下に示すスキーマを持つ customer という名前のテーブルを使用してプロセスを説明します。

1. ソーステーブルのスキーマに基づいて、マージされた結果を格納するためのテーブルを MaxCompute に作成します。

   たとえば、タイムスタンプ 1565944878 での customer テーブルの完全なデータを取得するには、識別しやすいように次のテーブルを作成します。

   CREATE TABLE `customer_1565944878` (
       `id` bigint NULL,
       `register_time` datetime NULL,
       `address` string);

   説明

   • MaxCompute の「アドホッククエリ」機能を使用して SQL コマンドを実行できます。

   • MaxCompute でサポートされているデータ型については、「データ型」をご参照ください。

2. MaxCompute で、次の SQL コマンドを実行して、全量ベースラインテーブルと増分ログテーブルをマージし、特定の時点での完全なデータを取得します。

   set odps.sql.allow.fullscan=true;
   insert overwrite table <result_storage_table>
   select <col1>,
          <col2>,
          <colN>
     from(
   select row_number() over(partition by t.<primary_key_column>
    order by record_id desc, after_flag desc) as row_number, record_id, operation_flag, after_flag, <col1>, <col2>, <colN>
     from(
   select incr.record_id, incr.operation_flag, incr.after_flag, incr.<col1>, incr.<col2>,incr.<colN>
     from <table_log> incr
    where utc_timestamp< <timestamp>
    union all
   select 0 as record_id, 'I' as operation_flag, 'Y' as after_flag, base.<col1>, base.<col2>,base.<colN>
     from <table_base> base) t) gt
   where row_number=1 
     and after_flag='Y'

   説明

   • <result_storage_table>：マージされた結果セットを格納するテーブルの名前。

   • <col1>/<col2>/<colN>：同期されたテーブルの列名。

   • <primary_key_column>：同期されたテーブルのプライマリキー列の名前。

   • <table_log>：増分ログテーブルの名前。

   • <table_base>：全量ベースラインテーブルの名前。

   • <timestamp>：完全なデータを取得したい時点。

   次の例は、データテーブルをマージして、タイムスタンプ 1565944878 での customer テーブルの完全なデータを取得する方法を示しています。

   set odps.sql.allow.fullscan=true;
   insert overwrite table customer_1565944878
   select id,
          register_time,
          address
     from(
   select row_number() over(partition by t.id
    order by record_id desc, after_flag desc) as row_number, record_id, operation_flag, after_flag, id, register_time, address
     from(
   select incr.record_id, incr.operation_flag, incr.after_flag, incr.id, incr.register_time, incr.address
     from customer_log incr
    where utc_timestamp< 1565944878
    union all
   select 0 as record_id, 'I' as operation_flag, 'Y' as after_flag, base.id, base.register_time, base.address
     from customer_base base) t) gt
    where gt.row_number= 1
      and gt.after_flag= 'Y';

3. コマンドが完了したら、customer_1565944878 テーブルでマージされたデータをクエリできます。