Hologres:MaxCompute のパーティションテーブルからのデータのインポート

注意事項

• 原子性を確保するために一時テーブルを使用します。データは、インポートの完了後にのみパーティションテーブルにアタッチされます。これにより、インポートタスクが失敗した場合でも、テーブルの削除などの手動クリーンアップ作業が不要になります。

• 子パーティション内のデータを更新する場合は、トランザクション整合性を確保するために、単一のトランザクション内で古い子テーブルを削除し、新しい一時テーブルをアタッチする必要があります。

• 移行アシスタントを使用して DataWorks タスクをインポートするには、次の条件を満たす必要があります：

  • DataWorks のエディションが Standard Edition 以降であること。詳細については、「DataWorks editions」をご参照ください。

  • DataWorks ワークスペースが MaxCompute および Hologres データソースにバインドされていること。詳細については、「Configure a workspace」をご参照ください。

  • DataWorks ワークスペースが従来の DataStudio を引き続き使用していること。移行アシスタントは、新しい Data Studio にアップグレードされたワークスペースをサポートしていません。アップグレード済みのワークスペースでは、リソースパッケージのインポート時に検証が失敗し、The migration assistant does not support workspaces that use the new Data Studio というメッセージが表示されます。その場合は、このトピックの「操作手順」に従ってタスクを手動で作成してください。

操作手順

1. MaxCompute データの準備

   1. DataWorks コンソールにログインし、ターゲットリージョンに切り替え、左側のナビゲーションウィンドウで データ分析とサービス > データ分析 を選択します。[データ分析へ] をクリックします。左側のナビゲーションウィンドウで、 アイコンをクリックして SQL クエリ ページに移動します。

   2. [SQL Query] ページで、次の SQL 文を入力してパーティションテーブルを作成し、実行 をクリックします。

      DROP TABLE IF EXISTS odps_sale_detail;
      -- sale_detail という名前のパーティションテーブルを作成します。
      CREATE TABLE IF NOT EXISTS odps_sale_detail 
      (
          shop_name STRING
          ,customer_id STRING
          ,total_price DOUBLE
      )
      PARTITIONED BY 
      (
          sale_date STRING
      )
      ;

   3. [SQL Query] ページで、次の SQL 文を入力してパーティションテーブルにデータをインポートし、実行 をクリックします。

      -- ソーステーブルに 20210815 パーティションを追加します。
      ALTER TABLE odps_sale_detail ADD IF NOT EXISTS PARTITION(sale_date='20210815')
      ;
      -- パーティションにデータを書き込みます。
      INSERT OVERWRITE TABLE odps_sale_detail PARTITION(sale_date='20210815') VALUES 
      ('s1','c1',100.1),
      ('s2','c2',100.2),
      ('s3','c3',100.3)
      ;
      -- ソーステーブルに 20210816 パーティションを追加します。
      ALTER TABLE odps_sale_detail ADD IF NOT EXISTS PARTITION(sale_date='20210816')
      ;
      -- パーティションにデータを書き込みます。
      INSERT OVERWRITE TABLE odps_sale_detail PARTITION(sale_date='20210816') VALUES 
      ('s1','c1',100.1),
      ('s2','c2',100.2),
      ('s3','c3',100.3)
      ;
      -- ソーステーブルに 20210817 パーティションを追加します。
      ALTER TABLE odps_sale_detail ADD IF NOT EXISTS PARTITION(sale_date='20210817')
      ;
      -- パーティションにデータを書き込みます。
      INSERT OVERWRITE TABLE odps_sale_detail PARTITION(sale_date='20210817') VALUES 
      ('s1','c1',100.1),
      ('s2','c2',100.2),
      ('s3','c3',100.3)
      ;
      -- ソーステーブルに 20210818 パーティションを追加します。
      ALTER TABLE odps_sale_detail ADD IF NOT EXISTS PARTITION(sale_date='20210818')
      ;
      -- パーティションにデータを書き込みます。
      INSERT OVERWRITE TABLE odps_sale_detail PARTITION(sale_date='20210818') VALUES 
      ('s1','c1',100.1),
      ('s2','c2',100.2),
      ('s3','c3',100.3)
      ;

2. Hologres でのテーブル作成

   • 外部テーブルの作成

     1. データベースにログインします。

        1. HoloWeb 開発ページで、メタデータ管理 をクリックします。

        2. [メタデータ管理] ページで、左側のディレクトリツリーにある作成済みデータベースの名前をダブルクリックし、その後、[OK] をクリックします。

     2. 外部テーブルの作成

        1. SQL エディター ページで、左上の アイコンをクリックして新しい SQL クエリを作成します。

        2. 新しい 一時クエリ ページで、インスタンス名 と データベース を選択します。SQL エディターに次の文を入力し、実行 をクリックします。

           DROP FOREIGN TABLE IF EXISTS odps_sale_detail;
           -- 外部テーブルを作成します。
           IMPORT FOREIGN SCHEMA maxcompute_project LIMIT to
           (
               odps_sale_detail
           ) 
           FROM SERVER odps_server INTO public 
           OPTIONS(if_table_exist 'error',if_unsupported_type 'error');

   • パーティションテーブル (内部テーブル) の作成

     1. データベースにログインします。

        1. HoloWeb 開発ページで、メタデータ管理 をクリックします。

        2. [メタデータ管理] ページで、左側のディレクトリツリーにある作成済みデータベースの名前をダブルクリックし、その後、[OK] をクリックします。

     2. パーティションテーブルの作成

        1. SQL エディター ページで、左上の アイコンをクリックして新しい SQL クエリを作成します。

        2. 新しい 一時クエリ ページで、インスタンス名 と データベース を選択します。SQL エディターに次の文を入力し、実行 をクリックします。

           DROP TABLE IF EXISTS holo_sale_detail;
           -- Hologres パーティションテーブル (内部テーブル) を作成します。
           BEGIN ;
           CREATE TABLE IF NOT EXISTS holo_sale_detail
           (
               shop_name TEXT
               ,customer_id TEXT 
               ,total_price FLOAT8
               ,sale_date TEXT
           )
           PARTITION BY LIST(sale_date);
           COMMIT;

3. Hologres 一時テーブルへのパーティションデータのインポート

   一時クエリ ページで、SQL エディターに次の文を入力し、実行 をクリックします。

   この SQL 文は、MaxCompute の hologres_test プロジェクトにある odps_sale_detail パーティションテーブルの 20210816 パーティションを、Hologres の holo_sale_detail パーティションテーブルの 20210816 パーティションにインポートします。

   説明

   Hologres V2.1.17 以降ではサーバーレスコンピューティングがサポートされています。大規模なオフラインデータインポート、大規模 ETL タスク、外部テーブルに対する高ボリュームのクエリなどのシナリオでは、サーバーレスコンピューティングを使用してこれらのタスクを実行できます。この機能は、インスタンスリソースの代わりに追加のサーバーレスリソースを使用するため、インスタンスの安定性が向上し、メモリ不足 (OOM) エラーの発生確率が低下します。インスタンス用に追加のコンピューティングリソースを予約する必要はなく、実行したタスクに対してのみ課金されます。サーバーレスコンピューティングの詳細については、「Serverless Computing」をご参照ください。サーバーレスコンピューティングの使用方法については、「Use Serverless Computing」をご参照ください。

   -- 一時テーブルが残っている可能性があるためクリーンアップします。
   BEGIN ;
   DROP TABLE IF EXISTS holo_sale_detail_tmp_20210816;
   COMMIT ;
   -- 一時テーブルを作成します。
   SET hg_experimental_enable_create_table_like_properties=on;
   BEGIN ;
   CALL HG_CREATE_TABLE_LIKE ('holo_sale_detail_tmp_20210816', 'select * from holo_sale_detail'); 
   COMMIT;
   -- (オプション) 大規模なオフラインデータインポートおよび ETL タスクは、サーバーレスコンピューティングで実行することを推奨します。
   SET hg_computing_resource = 'serverless';
   -- 一時テーブルにデータを挿入します。
   INSERT INTO holo_sale_detail_tmp_20210816
   SELECT *
   FROM public.odps_sale_detail
   WHERE sale_date='20210816';
   -- 必須ではない SQL クエリがサーバーレスリソースを使用しないよう、設定をリセットします。
   RESET hg_computing_resource;

4. Hologres パーティションテーブルへの一時テーブルのアタッチ

   一時クエリ ページで、SQL エディターに次の文を入力し、実行 をクリックします。

   • 古い子テーブルが存在する場合は、先に削除してから、一時テーブルを Hologres パーティションテーブルにアタッチする必要があります。

     この SQL 文は、子テーブル holo_sale_detail_20210816 を削除し、一時テーブル holo_sale_detail_tmp_20210816 を holo_sale_detail パーティションテーブルの 20210816 パーティションにアタッチします。

     -- 子テーブルが存在する場合は置き換えます。
     BEGIN ;
     -- 古い子テーブルを削除します。
     DROP TABLE IF EXISTS holo_sale_detail_20210816;
     -- 一時テーブルの名前を変更します。
     ALTER TABLE holo_sale_detail_tmp_20210816 RENAME TO holo_sale_detail_20210816;
     -- 一時テーブルを指定したパーティションテーブルにアタッチします。
     ALTER TABLE holo_sale_detail ATTACH PARTITION holo_sale_detail_20210816
     FOR VALUES IN ('20210816')
     ;
     COMMIT ;

   • 古い子テーブルが存在しない場合は、一時テーブルを直接 Hologres パーティションテーブルにアタッチします。

     この SQL 文は、一時テーブル holo_sale_detail_tmp_20210816 を、パーティションテーブル holo_sale_detail の 20210816 パーティションにアタッチします。

     BEGIN ;
     -- 一時テーブルの名前を変更します。
     ALTER TABLE holo_sale_detail_tmp_20210816 RENAME TO holo_sale_detail_20210816;
     -- 一時テーブルを指定したパーティションテーブルにアタッチします。
     ALTER TABLE holo_sale_detail ATTACH PARTITION holo_sale_detail_20210816
     FOR VALUES IN ('20210816');
     COMMIT ;

5. Hologres パーティションテーブルに対する ANALYZE の実行

   一時クエリ ページで、SQL エディターに次の文を入力し、実行 をクリックします。

   この SQL 文は、holo_sale_detail パーティションテーブルに対して ANALYZE を実行して統計情報を収集します。これにより、クエリプランナが効率的な実行計画を生成しやすくなります。パーティションテーブルに対して ANALYZE を実行する場合は、親テーブルに対してのみ実行すれば十分です。

   -- 大量データのインポート後に、パーティションテーブルの親テーブルに対して ANALYZE を実行します。
   ANALYZE holo_sale_detail;

6. 期限切れ子パーティションのクリーンアップ (オプション)

   本番環境ではデータにライフサイクルがあるため、保持期間を過ぎたパーティションをクリーンアップする必要があります。

   一時クエリ ページで、SQL エディターに次の文を入力し、実行 をクリックします。

   この SQL 文は、20210630 パーティションをクリーンアップします。

   DROP TABLE IF EXISTS holo_sale_detail_20210630;

フル初期化と増分更新

HoloWeb の [one-click MaxCompute data import] では、[Partition Field] ドロップダウンと 1 つの [Business Date] セレクターのみが提供されるため、一度に 1 つのパーティションしかインポートできず、すべての過去パーティションをまとめてインポートできません。前述の手順では、Hologres に MaxCompute の 外部テーブル を作成し、SQL 文 を使用してデータを 一時テーブル に書き込み (これによりインポートの 原子性 を保証)、その後、その一時テーブルをパーティションテーブル (内部テーブル) にアタッチします。別の方法として、SQL を記述せずに DataWorks の Data Integration で直接データを同期することもできます。すべての過去パーティションデータをインポートし、日次の増分更新を維持するには、次の 2 つのフェーズに従ってください。

開始する前に、DataWorks の ワークスペースが MaxCompute と Hologres の両方のデータソースにバインドされていることを確認してください。詳細については、このトピックの「注意事項」セクションをご参照ください。

1. フル初期化：HoloWeb の [console] で、[Data Migration and Synchronization] エントリから DataWorks の Data Integration に移動します。バッチ同期タスク を作成します。ソースとして MaxCompute パーティションテーブルを選択し、宛先としてターゲットの Hologres テーブルを選択します。フィルター条件を 1=1 に設定して、すべての過去パーティションデータを 1 回の操作でインポートします。この同期タスクを送信して実行し、フル初期化を完了します。所要時間は、過去パーティションの データ量 によって異なります。インポート完了後、パーティションテーブルの親テーブルに対して ANALYZE を実行し、パフォーマンス統計 を更新します。

2. 増分更新：本番環境 では、定期スケジュールによって新しいパーティションが継続的にインポートされます。bizdate パラメーターを使用して、この同期タスクに 定期スケジューリング を設定し、毎日その日のパーティションのみをインポートします。開始前にタスクを公開する必要があり、ステータスは [execution details] で確認できます。より包括的なスケジューリングのジョブオーケストレーション (パラメーターテーブルや期限切れパーティションのクリーンアップを含む) については、次の「DataWorks タスクの例」セクションをご参照ください。

DataWorks タスクの例

ワークフローモジュール

• 基本パラメーター

  基本パラメーターは、ワークフローで使用されるすべてのパラメーターを管理します。主なパラメーターは次のとおりです：

  Id	Parameter	Type	Value	Description
  1	datepre31	variable	${yyyymmdd-31}	期限切れパーティションのクリーンアップを制御するパラメーターです。この値は、31 日前のパーティションをクリーンアップすることを指定します。
  2	datetime1	variable	$bizdate	パーティション作成を制御するパラメーターです。
  3	holo_table_name	constant	holo_sale_detail	Hologres パーティションテーブルの名前です。
  4	odps_project	constant	hologres_test	MaxCompute プロジェクトの名前です。
  5	odps_table_name	constant	odps_sale_detail	MaxCompute パーティションテーブルの名前です。
  6	partition_key	constant	sale_date	MaxCompute テーブルのパーティションキーです。

• パーティションデータの一時テーブルへの書き込み

  この手順は Hologres SQL モジュールです。SQL コードは次のとおりです：

  説明

  Hologres V2.1.17 以降ではサーバーレスコンピューティングがサポートされています。大規模なオフラインデータインポート、大規模 ETL タスク、外部テーブルに対する高ボリュームのクエリなどのシナリオでは、サーバーレスコンピューティングを使用してこれらのタスクを実行できます。この機能は、インスタンスリソースの代わりに追加のサーバーレスリソースを使用するため、インスタンスの安定性が向上し、メモリ不足 (OOM) エラーの発生確率が低下します。インスタンス用に追加のコンピューティングリソースを予約する必要はなく、実行したタスクに対してのみ課金されます。サーバーレスコンピューティングの詳細については、「Serverless Computing」をご参照ください。サーバーレスコンピューティングの使用方法については、「Use Serverless Computing」をご参照ください。

  -- 一時テーブルが残っている可能性があるためクリーンアップします。
  BEGIN ;
  DROP TABLE IF EXISTS ${holo_table_name}_tmp_${datetime1};
  COMMIT ;
  -- 一時テーブルを作成します。
  SET hg_experimental_enable_create_table_like_properties=on;
  BEGIN ;
  CALL HG_CREATE_TABLE_LIKE ('${holo_table_name}_tmp_${datetime1}', 'select * from ${holo_table_name}'); 
  COMMIT;
  -- (オプション) 大規模なオフラインデータインポートおよび ETL タスクは、サーバーレスコンピューティングで実行することを推奨します。
  SET hg_computing_resource = 'serverless';
  -- 一時テーブルにデータを挿入します。
  INSERT INTO ${holo_table_name}_tmp_${datetime1}
  SELECT *
  FROM public.${odps_table_name}
  WHERE ${partition_key}='${datetime1}';
  -- 必須ではない SQL クエリがサーバーレスリソースを使用しないよう、設定をリセットします。
  RESET hg_computing_resource;

  このモジュールの上流入力として基本パラメーターをバインドする必要があります。このノードの [Node Context] 設定で、[Input Parameters of This Node] セクションには 5 つのパラメーター datetime1、holo_table_name、odps_project、odps_table_name、partition_key が含まれます。すべてのパラメーターの値ソースは上流ノード holo_doc_test.501890067_out です。手動で追加してください。[Output Parameters of This Node] セクションは空です。

• 子テーブルの置き換え

  この手順は、既存の子テーブルを置き換える Hologres SQL モジュールです。原子性を確保するため、操作はトランザクション内で実行されます。SQL コードは次のとおりです：

  -- 子テーブルが存在する場合は置き換えます。
  BEGIN ;
  -- 既存の子テーブルを削除します。
  DROP TABLE IF EXISTS ${holo_table_name}_${datetime1};
  -- 一時テーブルの名前を変更します。
  ALTER TABLE ${holo_table_name}_tmp_${datetime1} RENAME TO ${holo_table_name}_${datetime1};
  -- 一時テーブルを指定したパーティションテーブルにアタッチします。
  ALTER TABLE ${holo_table_name} ATTACH PARTITION ${holo_table_name}_${datetime1}
  FOR VALUES IN ('${datetime1}');
  COMMIT ;

  このモジュールの上流入力として基本パラメーターをバインドする必要があります。[Node Context] の [Input Parameters of This Node] で、パラメーター datetime1 と holo_table_name を手動で追加します。値ソースは、それぞれ上流ノードの出力 hologres_doc_test.502282431_out:datetime1 と hologres_doc_test.502282431_out:holo_table_name に設定します。

• パーティションテーブルの統計情報の収集

  この手順は、親テーブルの統計情報を収集する Hologres SQL モジュールです。SQL コードは次のとおりです：

  -- 大量データのインポート後に、パーティションテーブルの親テーブルに対して ANALYZE を実行します。
  ANALYZE ${holo_table_name};

  このモジュールの上流入力として基本パラメーターをバインドする必要があります。[Node Context] の [Input Parameters of This Node] で [Add] をクリックし、パラメーター holo_table_name を手動で追加します。値ソースを holo_doc_test.501890067_out:holo_table_name に設定します。

• 期限切れ子テーブルのクリーンアップ

  本番環境ではデータにライフサイクルがあるため、保持期間を過ぎたパーティションをクリーンアップする必要があります。

  この例では、Hologres に直近 31 日分のパーティションのみを保持する方法を示します。パラメーターが datepre31=${yyyymmdd-31} に設定されているため、期限切れ子テーブルを削除する SQL コードは次のとおりです：物理パーティションテーブルの子テーブルを削除するには、まず親テーブルからデタッチ (DETACH PARTITION) し、その後に DROP TABLE を実行する必要があります。そうしないと、子テーブルが親テーブルにアタッチされたままになります。

  -- 期限切れ子テーブルのクリーンアップ：最初に子パーティションをデタッチし、その後テーブルを削除します。
  BEGIN ;
  ALTER TABLE ${holo_table_name} DETACH PARTITION ${holo_table_name}_${datepre31};
  DROP TABLE IF EXISTS ${holo_table_name}_${datepre31};
  COMMIT ;

  たとえば、bizdate が 20200309 の場合、datepre31 は 20200207 になります。これにより、パーティションのクリーンアップ日付が正しく設定されます。

  このモジュールでも、基本パラメーターを上流入力としてバインドする必要があります。[Node Context] パネルの [Input Parameters of This Node] セクションで [Add] をクリックし、次の 2 つのパラメーターを手動で追加します。パラメーター名 datepre31 の値ソースを上流ノード出力 holo_doc_test.501890067_out:datepre31 に設定します。パラメーター名 holo_table_name の値ソースを上流ノード出力 holo_doc_test.501890067_out:holo_table_name に設定します。

移行アシスタントを使用した DataWorks タスクのインポート

• タスクが複雑であるため、DataWorks の移行アシスタントを使用して、提供されているタスクパッケージをプロジェクトにインポートできます。これにより、このトピックで説明する事前設定済みの DataWorks タスクが提供され、パラメーターやスクリプトを変更して特定のビジネス要件に合わせてカスタマイズできます。

• 次のタスクパッケージをダウンロードします：DataWorks job package。

1. DataWorks の移行アシスタントを開きます。詳細については、「Open the migration assistant」をご参照ください。

2. 移行アシスタントの左側メニューで、DataWorks の移行 > DataWorks インポート をクリックします。

3. DataWorks インポート ページで、右上の インポートタスクの新規作成 をクリックします。

4. インポートタスクの新規作成 ダイアログボックスで、パラメーターを設定します。

   Parameter	Description
   インポート名	任意の名前です。インポート名には、大文字、小文字、日本語や中国語の文字、数字、アンダースコア (_) およびピリオド (.) を使用できます。
   アップロード方式	ファイルをアップロードする方法です： ローカルアップロード：この方法を使用して、30 MB 以下のエクスポートパッケージファイルを DataWorks ワークスペースにアップロードします。 OSS リンク：エクスポートパッケージファイルが 30 MB を超える場合は、ファイルを OSS にアップロードし、DataWorks ワークスペースにファイル URL を入力します。ファイル URL を取得するには、OSS コンソールのファイル詳細ページに移動し、[File Url] セクションで URL をコピーします。ファイルのアップロード方法の詳細については、「Upload files」をご参照ください。OSS のダウンロードリンクを取得する方法の詳細については、「Share files」をご参照ください。
   備考	インポートタスクの簡単な説明です。

5. OK をクリックして インポートタスクの設定 ページに移動し、マッピングを設定します。

6. 左下の インポートの開始 をクリックし、確認 ダイアログボックスで OK をクリックします。

   1. インポートが成功すると、上記のスケジュール済みタスクが [Data Development] モジュールに表示されます。

   2. 関連する DDL 文も手動ワークフローに表示されます。