DataWorks を使用した MaxCompute から Hologres へのデータインポート - Hologres

DataWorks のスケジューリングタスクを使用して、MaxCompute のパーティションテーブルから Hologres のパーティションテーブルにデータを定期的にインポートできます。これにより、データの整合性と精度が確保され、両プラットフォームの強みが組み合わさり、データ処理の効率と信頼性が向上します。

前提条件

Hologres インスタンスを購入して有効化します。詳細については、「Hologres インスタンスの購入」をご参照ください。
MaxCompute を有効化し、プロジェクトを作成します。詳細については、「MaxCompute の有効化」をご参照ください。
DataWorks を有効化し、ワークスペースを作成します。詳細については、「ワークスペースの作成」をご参照ください。

注意事項

MaxCompute のテナントまたはプロジェクトでスキーマサービスが有効になっていないことを確認してください。スキーマの詳細については、「スキーマ操作」をご参照ください。

データ準備

このトピックでは、MaxCompute のパブリックデータセット `public_data` のパーティションテーブル `dwd_product_movie_basic_info` を例として使用します。次のコードは、`dwd_product_movie_basic_info` テーブルのスキーマを示しています。

-- MaxCompute パーティションテーブルの DDL
CREATE TABLE IF NOT EXISTS public_data.dwd_product_movie_basic_info(
  movie_name STRING COMMENT '映画名。',
  director STRING COMMENT '監督。',
  scriptwriter STRING COMMENT '脚本家。',
  area STRING COMMENT '制作地域または国。',
  actors STRING COMMENT '出演者。',
  `type` STRING COMMENT 'ジャンル。',
  movie_length STRING COMMENT '映画の長さ。',
  movie_date STRING COMMENT '公開日。',
  movie_language STRING COMMENT '言語。',
  imdb_url STRING COMMENT 'IMDB ID。'
) 
PARTITIONED BY (ds STRING) STORED AS ALIORC;

操作手順

MaxCompute データの準備
1. DataWorks コンソールにログインし、ターゲットリージョンに切り替え、左側のナビゲーションウィンドウで データ分析とサービス > Data Analysis を選択します。[データ分析へ] をクリックします。左側のナビゲーションウィンドウで、アイコンをクリックして SQL Query ページに移動します。
2. SQL Query ページで、次の SQL ステートメントを入力して 20170112 パーティションのデータを表示し、Run をクリックします。
```
SELECT * FROM public_data.dwd_product_movie_basic_info WHERE ds = '20170112';
```
Hologres での外部テーブルの作成
外部テーブルの列の順序とデータ型は、MaxCompute テーブルのものと一致している必要があります。
1. HoloWeb SQL エディターページにログインします。
2. SQL Editor タブの下にあるアイコンをクリックして、Ad-hoc Query ウィンドウを開きます。クエリウィンドウのツールバーで、ご利用の Hologres インスタンスの Instance Name と データベース を選択します。
3. Ad-hoc Query エディターで、次の文を入力して Run をクリックします。
  次の文では、IMPORT FOREIGN SCHEMA コマンドを使用して、`dwd_product_movie_basic_info` という名前の Hologres 外部テーブルを作成します。
```
IMPORT FOREIGN SCHEMA public_data LIMIT TO (dwd_product_movie_basic_info) FROM SERVER odps_server INTO public OPTIONS(if_table_exist 'update');
```

Hologres での内部テーブルの作成

データを受信して保存するために、Hologres に内部テーブルを作成します。

HoloWeb 開発ページで、Ad-hoc Query をクリックします。

新しい Ad-hoc Query ウィンドウで、ご利用の Instance Name と データベース を選択します。次に、SQL エディターに次の文を入力し、Run をクリックします。

この例では MaxCompute のパーティションテーブルからデータをインポートするため、Hologres のターゲット内部テーブルもパーティションテーブルである必要があります。

説明

次の DDL 文は簡単な例です。最適なクエリパフォーマンスを得るには、実際のビジネス要件に基づいてテーブルを作成し、適切なインデックスを設定してください。

BEGIN;
CREATE TABLE "public"."holo_dwd_product_movie_basic_info" (
 "movie_name" TEXT,
 "director" TEXT,
 "scriptwriter" TEXT,
 "area" TEXT,
 "actors" TEXT,
 "type" TEXT,
 "movie_length" TEXT,
 "movie_date" TEXT,
 "movie_language" TEXT,
 "imdb_url" TEXT,
 "ds" TEXT
)
PARTITION BY LIST (ds);
CALL SET_TABLE_PROPERTY('"public"."holo_dwd_product_movie_basic_info"', 'orientation', 'column');
COMMENT ON COLUMN "public"."holo_dwd_product_movie_basic_info"."movie_name" IS '映画名。';
COMMENT ON COLUMN "public"."holo_dwd_product_movie_basic_info"."director" IS '監督。';
COMMENT ON COLUMN "public"."holo_dwd_product_movie_basic_info"."scriptwriter" IS '脚本家。';
COMMENT ON COLUMN "public"."holo_dwd_product_movie_basic_info"."area" IS '制作地域または国。';
COMMENT ON COLUMN "public"."holo_dwd_product_movie_basic_info"."actors" IS '出演者。';
COMMENT ON COLUMN "public"."holo_dwd_product_movie_basic_info"."type" IS 'ジャンル。';
COMMENT ON COLUMN "public"."holo_dwd_product_movie_basic_info"."movie_length" IS '映画の長さ。';
COMMENT ON COLUMN "public"."holo_dwd_product_movie_basic_info"."movie_date" IS '公開日。';
COMMENT ON COLUMN "public"."holo_dwd_product_movie_basic_info"."movie_language" IS '言語。';
COMMENT ON COLUMN "public"."holo_dwd_product_movie_basic_info"."imdb_url" IS 'IMDB ID。';
COMMIT;

データインポートタスクの開発

このステップでは、Hologres SQL ノードを作成して、パーティションテーブルでスケジュールされたジョブを実行します。

DataWorks コンソールにログインし、データ開発ページに移動して、Hologres SQL ノードを作成します。詳細については、「Hologres SQL ノード」をご参照ください。

ノードエディターで、次の文を入力します。

Hologres では、パーティション分割されたデータを親パーティションテーブルに直接書き込むことはできません。まず、ソースの MaxCompute パーティションテーブルのパーティションキー値に対応する子パーティションテーブルを Hologres に作成する必要があります。その後、パーティション分割されたデータを対応する子パーティションテーブルにインポートできます。パーティションキー値は ${bizdate} パラメーターによって制御され、スケジューリングシステムによって自動的に値が割り当てられ、定期的なスケジューリングが可能になります。スケジューリングパラメーターの詳細については、「サポートされているスケジューリングパラメーターのフォーマット」をご参照ください。

説明

インポートされるパーティションデータは、パーティションキー値 (この例では ds) と一致する必要があります。一致しない場合、エラーが発生します。

次の 2 つのシナリオは、異なるアプローチを示しています。ビジネスロジックに基づいていずれかを選択してください。

シナリオ 1：新しいパーティションデータのインポート

-- 一時的な子パーティションテーブルを作成します。
BEGIN;
CREATE TABLE IF NOT EXISTS "public".tmp_holo_dwd_product_movie_basic_info_${bizdate}  (
 "movie_name" TEXT,
 "director" TEXT,
 "scriptwriter" TEXT,
 "area" TEXT,
 "actors" TEXT,
 "type" TEXT,
 "movie_length" TEXT,
 "movie_date" TEXT,
 "movie_language" TEXT,
 "imdb_url" TEXT,
 "ds" TEXT
);
COMMIT;
-- 外部テーブルのスキーマを更新します。
IMPORT FOREIGN SCHEMA public_data LIMIT TO (dwd_product_movie_basic_info) FROM SERVER odps_server INTO public OPTIONS(if_table_exist 'update');
-- Hologres のメタデータキャッシュの更新が遅いために同期が失敗するのを防ぐため、データをインポートする前に 30 秒待機します。
SELECT pg_sleep(30); 
-- MaxCompute から一時的な子パーティションテーブルにデータをインポートします。
INSERT INTO "public".tmp_holo_dwd_product_movie_basic_info_${bizdate} 
SELECT 
    "movie_name",
    "director",
    "scriptwriter",
    "area",
    "actors",
    "type",
    "movie_length",
    "movie_date",
    "movie_language",
    "imdb_url",
    "ds"
FROM "public".dwd_product_movie_basic_info
WHERE ds='${bizdate}';
-- 新しいパーティションデータをインポートします。
BEGIN;
ALTER TABLE tmp_holo_dwd_product_movie_basic_info_${bizdate} RENAME TO holo_dwd_product_movie_basic_info_${bizdate};
-- 新しい子パーティションテーブルを親パーティションテーブルにアタッチします。
ALTER TABLE holo_dwd_product_movie_basic_info ATTACH PARTITION holo_dwd_product_movie_basic_info_${bizdate} FOR VALUES IN ('${bizdate}');
COMMIT;

シナリオ 2：履歴パーティションデータの更新

-- 一時的な子パーティションテーブルを作成します。
BEGIN;
CREATE TABLE IF NOT EXISTS "public".tmp_holo_dwd_product_movie_basic_info_${bizdate}  (
 "movie_name" TEXT,
 "director" TEXT,
 "scriptwriter" TEXT,
 "area" TEXT,
 "actors" TEXT,
 "type" TEXT,
 "movie_length" TEXT,
 "movie_date" TEXT,
 "movie_language" TEXT,
 "imdb_url" TEXT,
 "ds" TEXT
);
COMMIT;
-- 外部テーブルのスキーマを更新します。
IMPORT FOREIGN SCHEMA public_data LIMIT TO (dwd_product_movie_basic_info) FROM SERVER odps_server INTO public OPTIONS(if_table_exist 'update');
-- Hologres のメタデータキャッシュの更新が遅いために同期が失敗するのを防ぐため、データをインポートする前に 30 秒待機します。
SELECT pg_sleep(30); 
-- MaxCompute から一時的な子パーティションテーブルにデータをインポートします。
INSERT INTO "public".tmp_holo_dwd_product_movie_basic_info_${bizdate} 
SELECT 
    "movie_name",
    "director",
    "scriptwriter",
    "area",
    "actors",
    "type",
    "movie_length",
    "movie_date",
    "movie_language",
    "imdb_url",
    "ds"
FROM "public".dwd_product_movie_basic_info
WHERE ds='${bizdate}';
-- 履歴パーティションデータを更新します。
BEGIN;
ALTER TABLE IF EXISTS holo_dwd_product_movie_basic_info DETACH PARTITION holo_dwd_product_movie_basic_info_${bizdate};
DROP TABLE IF EXISTS holo_dwd_product_movie_basic_info_${bizdate};
ALTER TABLE tmp_holo_dwd_product_movie_basic_info_${bizdate} RENAME TO holo_dwd_product_movie_basic_info_${bizdate};
-- 子パーティションテーブルを親パーティションテーブルにアタッチします。
ALTER TABLE holo_dwd_product_movie_basic_info ATTACH PARTITION holo_dwd_product_movie_basic_info_${bizdate} FOR VALUES IN ('${bizdate}');
COMMIT;

スケジューリングプロパティの構成
Hologres SQL ノードエディターページで、右側のペインにある Configure Scheduling をクリックして、ノードのスケジューリングプロパティを構成します。
説明
次のパラメーターを変更し、その他はデフォルト値のままにします。
- 基本プロパティ
  パラメーター
  値
  パラメーター
  bizdate=${yyyymmdd}
- 時間プロパティ：[時間プロパティ] セクションで、[スケジューリングタイプ] を [通常スケジューリング] に設定し、[有効期間] をデフォルト (1970-01-01 から 9999-01-01) のままにし、[スケジューリング周期] を [日] に設定し、[タイムアウト] を [システムデフォルト] に設定し、[前の周期に依存] は選択しません。
  パラメーター
  値
  インスタンス生成モード
  デプロイ直後
  再実行
  失敗時のみ許可
  スケジュール時刻
  00:05
- スケジューリングの依存関係
  ビジネスロジックに基づいて、依存関係をルートノードまたは既存の親ノードに設定できます。まず、[自動解析] を [オン] に設定し、次に [血縁関係の解析] をクリックします。システムは自動的にルートノードを解析します。その後、[自動解析] を [オフ] に設定します。
スケジューリングノードの公開
1. Hologres SQL ノードエディターページで、ツールバーのアイコンをクリックしてノードを保存します。
2. ツールバーのアイコンをクリックしてノードを送信します。
3. [新しいバージョンを送信] ダイアログボックスで、[変更の説明] を入力します。
4. OK をクリックします。
オペレーションセンターでのノードの実行
1. Hologres SQL ノードエディターページで、右上隅にある [O&M] をクリックします。
2. [オペレーションセンター] ページの左側のナビゲーションウィンドウで、[自動トリガーノードのメンテナンス] > Recurring Job を選択します。
3. Recurring Job ページで、ノードを右クリックし、[データ補填] > [現在のノード] を選択します。
4. 左側のナビゲーションウィンドウで、[アシスタント] > [データ補填] を選択して、実行中のタスクとそのステータスを表示します。
データの表示
タスクが正常に実行されると、Hologres はパーティションのデータ用に子パーティションテーブルを自動的に作成します。
1. データ開発ページで Hologres SQL ノードを作成します。詳細については、「Hologres SQL ノード」をご参照ください。
2. ノードエディターページで、次の文を入力してデータをクエリします。
  - 子パーティションテーブルのデータを表示します。
```
SELECT * FROM holo_dwd_product_movie_basic_info_20170112;
```
  - 親パーティションテーブルの総行数を表示します。
```
SELECT COUNT (*) FROM holo_dwd_product_movie_basic_info;
```

パラメーター	値
パラメーター	bizdate=${yyyymmdd}

パラメーター	値
インスタンス生成モード	デプロイ直後
再実行	失敗時のみ許可
スケジュール時刻	00:05