Data Transmission Service:PolarDB\-X 2\.0 から MaxCompute へのデータ同期

前提条件

開始する前に、以下の条件を満たしていることを確認してください。

• MySQL 5.7 と互換の PolarDB-X 2.0 インスタンスです。「インスタンスの作成」および「データベースの作成」を参照してください。

• MaxCompute の有効化。詳細については、「MaxCompute および DataWorks の有効化」をご参照ください。

• MaxCompute プロジェクトの作成。詳細については、「MaxCompute プロジェクトの作成」をご参照ください。

• ソース PolarDB-X インスタンス上のデータベースアカウント（以下の権限が必要）：

  権限	使用タイミング	必要性の理由
  SELECT	初期完全データ同期時	同期対象オブジェクトの行を読み取ります。
  REPLICATION CLIENT	常時	バイナリログのステータス（SHOW MASTER STATUS、SHOW BINARY LOGS）を照会します。
  REPLICATION SLAVE	増分データ同期	ソースインスタンスに接続し、バイナリログを読み取ります。

  これらの権限を付与する手順については、「PolarDB-X のデータ同期ツール」をご参照ください。

• PolarDB-X 2.0 コンソールでバイナリログ記録が有効化され、binlog_row_image が full に設定されています。「パラメーター設定」を参照してください。binlog_row_image が full に設定されていない場合、事前チェックが失敗し、タスクを開始できません。

• バイナリログの保存期間を以下のように設定します。

  同期範囲	最低保存期間
  増分同期のみ	24 時間
  全量＋増分同期	7 日間（全量同期完了後は 24 時間に短縮可能）

  DTS が保存期間が短すぎるためバイナリログを読み取れない場合、タスクは失敗し、データの不整合または損失が発生する可能性があります。上記要件に従って保存期間を設定してください。設定が適切でない場合、Data Transmission Service (DTS) のサービスレベル合意 (SLA) で定められたサービスの信頼性およびパフォーマンスを保証できません。

制限事項

外部キーは同期されません。 ソースデータベースにおける外部キー制約によってトリガーされる CASCADE および DELETE 操作は、MaxCompute にはレプリケーションされません。

課金

サポートされる SQL 操作

同期タスクの作成

増分データテーブルのスキーマ

MaxCompute プロジェクトで全表スキャンを許可するには、set odps.sql.allow.fullscan=true; を実行します。

DTS は PolarDB-X 2.0 からの増分変更を MaxCompute の増分データテーブルに書き込みます。ソーステーブルの列に加えて、DTS はすべての増分データテーブルに以下のメタデータ列を追加します。

この例では、modifytime_year、modifytime_month、modifytime_day、modifytime_hour、modifytime_minute が、ステップ 3 で設定された通りパーティションキーを構成します。

全量ベースラインテーブルと増分データテーブルのマージ

同期タスクが開始されると、DTS は MaxCompute 内に全量ベースラインテーブルおよび増分データテーブルを作成します。SQL を使用して 2 つのテーブルをマージし、任意の時点における完全なデータセットを再構築します。

以下の例では、customer という名前のテーブルのデータをマージします。

1. ソーステーブルと同じスキーマを持つ宛先テーブルを作成し、マージ結果を格納します。以下の例では、タイムスタンプ 1565944878 時点の customer データを格納するテーブルを作成します。

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

   アドホッククエリ機能を使用して、MaxCompute で SQL ステートメントを実行します。詳細については、「SQL ステートメントを実行するためのアドホッククエリ機能の使用」をご参照ください。サポートされているデータの型については、「データの型のエディション」をご参照ください。

2. 以下の 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 からマージされたデータをクエリします。