Flink SQLによるETLタスクの設定方法 - Data Transmission Service

Flink SQL は標準の SQL 構文を使用することで、ETL (抽出、変換、ロード) 開発を簡素化し、スキル習得のハードルを下げます。視覚的な DAG モードと比較して、Flink SQL はより強力で、DAG モードではサポートされていない高度な構文を記述できます。本トピックでは、Flink SQL を使用した ETL タスクの設定方法について説明します。

背景情報

説明

この機能は廃止予定であり、過去にご利用のあったユーザーのみ無料でご利用いただけます。新規ユーザーは本機能にアクセスできません。データ同期またはデータ移行インスタンス内で ETL タスクを設定することを推奨します。詳細については、「DTSのデータ移行または同期タスクにおけるETLタスクの設定」を参照してください。

ETL タスクを設定する前に、次の概念を理解しておく必要があります。
- [入力/ディメンションテーブル] は、ETL タスクのソースデータベースです。
- [出力] は、ETL で処理後のデータを書き込む宛先データベースです。
Data Transmission Service (DTS) は、データ同期プロセス内でストリーミング ETL 機能を提供しており、ソースデータベースと宛先データベースの間に変換コンポーネントを追加できます。これらのコンポーネントは、多彩なデータ変換を実行し、処理後のデータをリアルタイムで宛先に書き込みます。たとえば、2つのストリームテーブルを結合してワイドテーブルを作成し、それを宛先データベースに書き込むことができます。また、ソーステーブルに新しいフィールドを追加し、関数を使用してそのフィールドに値を割り当て、変換後のフィールドを宛先データベースに書き込むこともできます。

前提条件

ETL タスクは、次のリージョンでのみ利用可能です：China (Hangzhou)、China (Shanghai)、China (Qingdao)、China (Beijing)、China (Zhangjiakou)、China (Shenzhen)、China (Guangzhou)、China (Hong Kong)。
ソースデータベースは、次のいずれかのタイプである必要があります：[MySQL]、[PolarDB for MySQL]、[Oracle]、[PostgreSQL]、[DB2 for iSeries (AS/400)]、[DB2 for LUW]、[PolarDB-X (旧 DRDS)]、PolarDB for PostgreSQL、[MariaDB]、[PolarDB for Oracle]、[SQL Server]、または[PolarDB-X 2.0]。
ターゲットデータベースは、次のいずれかのタイプである必要があります：[MySQL]、[PolarDB for MySQL]、[Oracle]、[AnalyticDB for MySQL V3.0]、PolarDB for PostgreSQL、[PostgreSQL]、[DB2 for LUW]、[DB2 for iSeries (AS/400)]、[AnalyticDB for PostgreSQL]、[SQL Server]、[MariaDB]、[PolarDB-X (旧 DRDS)]、[PolarDB for Oracle]、または[Tablestore]。
ETL はスキーマ移行をサポートしていません。変換ロジックに基づいて、必要なテーブルスキーマをターゲットデータベースに作成してください。例えば、テーブル A (列 1、2、3) とテーブル B (列 2、3、4) を結合して列 2 と 3 を出力する場合は、まず列 2 と 3 を持つターゲットテーブル (テーブル C) を作成します。
ETL は全量データ同期をサポートしていません。増分データのみがリアルタイムで変換されます。

注意事項

ソースデータベースとターゲットデータベースは、同じリージョンにある必要があります。
すべての[ストリームテーブル]は、同じインスタンスのものである必要があります。
データベース名とテーブル名は一意である必要があります。
現在、クロスアカウントタスクはサポートされていません。

操作手順

ストリーミング ETL ページに移動します。
1. DTS コンソールにログインします。
2. 左側のナビゲーションペインで、ETL をクリックします。
左上隅にあるをクリックします。データフローの作成 ダイアログボックスで、データフロー名 フィールドに ETL タスクの名前を入力し、開発方法 を [FlinkSQL] に設定します。
確認をクリックします。

ETL のストリーミング ページの データフロー情報 セクションで、ソースデータベースと宛先データベースを追加します。

パラメーター	説明
リージョン	データソースのリージョンを選択します。
タイプ	テーブルタイプを選択します。ソーステーブルの設定では、ソースがストリームテーブル (リアルタイムで変化し、データエンリッチメントのためにディメンションテーブルと結合できるテーブル) の場合はストリームテーブルを、ソースがディメンションテーブル (更新頻度が低く、通常はリアルタイムデータと結合してワイドテーブルを作成するために使用されるテーブル) の場合はディメンションテーブルを選択します。宛先テーブルの設定では、出力を選択します。
データベースタイプ	ソースデータベースまたは宛先データベースのタイプを選択します。
インスタンス	名前またはIDを入力して、ソースインスタンスと宛先インスタンスを検索し、選択します。重要事前に、Data Management (DMS) にソースインスタンスと宛先インスタンスを追加しておく必要があります。手順については、「インスタンス管理」を参照してください。
データベース	変換対象のデータを含むソースデータベースまたは宛先データベースを選択します。
物理テーブル	変換対象のデータを含むソーステーブルまたは宛先テーブルを選択します。
物理テーブルのエイリアス	ソーステーブルまたは宛先テーブルに、簡潔で読みやすいエイリアスを設定します。このエイリアスは、ETL が SQL 文の実行時に正しいテーブルを特定するために使用されます。

ETL のストリーミング ページの SQL コマンドウィンドウで、SQL 文を追加して ETL タスクを設定します。

この例では、次の SQL 文を使用し、test_ordersストリームテーブルとproductディメンションテーブルを結合して、その結果をtest_orders_new宛先テーブルに挿入します。

重要

SQL 文はセミコロン (;) で区切る必要があります。

CREATE TABLE `etltest_test_orders` (
  `order_id` BIGINT,
  `user_id` BIGINT,
  `product_id` BIGINT,
  `total_price` DECIMAL(15,2),
  `order_date` TIMESTAMP(6),
  `dts_etl_schema_db_table` STRING,
  `dts_etl_db_log_time` BIGINT,
  `pt` AS PROCTIME(),
  WATERMARK FOR `order_date` AS `order_date` - INTERVAL '5' SECOND
) WITH (
  'streamType'= 'append',
  'alias'= 'test_orders',
  'vertexType'= 'stream'
);
CREATE TABLE `etltest_product` (
  `product_id` BIGINT,
  `product_name` STRING,
  `product_price` DECIMAL(15,2)
) WITH (
  'alias'= 'product',
  'vertexType'= 'lookup'
);
CREATE VIEW `etltest_test_orders_JOIN_etltest_product` AS
SELECT
  `etltest_test_orders`.`order_id` AS `order_id`,
  `etltest_test_orders`.`user_id` AS `user_id`,
  `etltest_test_orders`.`product_id` AS `product_id`,
  `etltest_test_orders`.`total_price` AS `total_price`,
  `etltest_test_orders`.`order_date` AS `order_date`,
  `etltest_test_orders`.`dts_etl_schema_db_table` AS `dts_etl_schema_db_table`,
  `etltest_test_orders`.`dts_etl_db_log_time` AS `dts_etl_db_log_time`,
  `etltest_product`.`product_id` AS `product_id_0001011101`,
  `etltest_product`.`product_name` AS `product_name`,
  `etltest_product`.`product_price` AS `product_price`
FROM `etltest_test_orders` LEFT JOIN `etltest_product` FOR SYSTEM_TIME AS OF `etltest_test_orders`.`pt` ON etltest_test_orders.product_id = etltest_product.product_id
;
CREATE TABLE `test_orders_new` (
  `order_id` BIGINT,
  `user_id` BIGINT,
  `product_id` BIGINT,
  `total_price` DECIMAL(15,2),
  `order_date` TIMESTAMP(6),
  `product_name` STRING,
  `product_price` DECIMAL(15,2)
) WITH (
  'alias'= 'test_orders_new',
  'vertexType'= 'sink'
);
INSERT INTO `test_orders_new` (
  `order_id`,
  `user_id`,
  `product_id`,
  `total_price`,
  `order_date`,
  `product_name`,
  `product_price`
)
SELECT
  `etltest_test_orders_JOIN_etltest_product`.`order_id`,
  `etltest_test_orders_JOIN_etltest_product`.`user_id`,
  `etltest_test_orders_JOIN_etltest_product`.`product_id`,
  `etltest_test_orders_JOIN_etltest_product`.`total_price`,
  `etltest_test_orders_JOIN_etltest_product`.`order_date`,
  `etltest_test_orders_JOIN_etltest_product`.`product_name`,
  `etltest_test_orders_JOIN_etltest_product`.`product_price`
FROM `etltest_test_orders_JOIN_etltest_product`;

タイプ	説明
ソーステーブルと宛先テーブルの情報	`CREATE TABLE` 文を使用して、ソーステーブルと宛先テーブルを定義します。 SQL 文の `WITH` 句では、`streamType`、`alias`、`vertexType` の3つのパラメーターを設定できます。ストリームテーブルの場合、3つすべてのパラメーターが必須です。ディメンションテーブルと出力テーブルの場合、`alias` と `vertexType` のみ必須です。 `streamType`：ストリームタイプ。データ処理中、ETL はストリームを動的テーブルに変換します。この動的テーブル上で連続クエリが実行され、`INSERT`、`UPDATE`、`DELETE` 操作によって絶えず変更され、新しい動的テーブルが生成されます。新しい動的テーブルが宛先データベースに書き込まれると、DTS はこのテーブルをストリームに戻します。このパラメーターは、動的テーブルがストリームに変換される際の変更をエンコードするために指定する必要があります。 Upsert：upsert ストリームです。動的テーブル内のデータは `INSERT`、`UPDATE`、および `DELETE` 操作によって変更可能です。ストリームに変換されると、`INSERT` および `UPDATE` 操作はupsertメッセージとしてエンコードされ、`DELETE` 操作はdeleteメッセージとしてエンコードされます。説明このエンコーディングでは、動的テーブルに一意のキーが必要です。このキーは複合キーでもかまいません。 append：アペンド専用ストリームです。動的テーブル内のデータは `INSERT` 操作によってのみ変更できます。ストリームに変換されると、挿入された行のみが出力されます。 `alias`：ステップ4でソースデータベースと宛先データベースに設定した物理テーブルのエイリアスを指します。 `vertexType`：テーブルタイプ。 `stream`：ストリーム (ストリームテーブル)。 `lookup`：ルックアップ (ディメンションテーブル)。 `sink`：シンク (宛先テーブル)。
データ変換ロジック	`CREATE VIEW` 文を使用して、データ変換ロジックを定義します。
変換後の宛先テーブルの情報	`INSERT INTO` 文を使用して、変換後の宛先テーブルにデータを書き込みます。

ソースデータベース、宛先データベースと SQL 文を設定した後、Flink SQL 検証の生成 をクリックします。
説明
- 公開をクリックして、直接、検証と事前チェックを実行することもできます。
- Flink SQL 検証が成功した場合、をクリックして検証の詳細を表示できます。
- Flink SQL 検証が失敗した場合、をクリックし、プロンプトメッセージに基づいて SQL 文を修正し、Flink SQL 検証を再実行してください。
Flink SQL 検証が成功したら、公開をクリックして事前チェックを開始します。
事前チェックに合格したら、[次へ：購入 (無料)] をクリックします。

説明
事前チェックが失敗した場合は、失敗した項目の横にある 詳細を表示 をクリックし、表示された詳細に基づいて問題を修正してから、事前チェックを再実行してください。
購入ページで、インスタンスクラス を選択します。[計算ユニット (CU)] の数は、パブリックプレビュー中は [2] に固定されています。[「Data Transmission Service (Pay-as-you-go) 利用規約」]および[「パブリックプレビュー利用規約」]を読み、同意のうえチェックボックスを選択します。

説明
パブリックプレビュー期間中、各ユーザーは無料で2つのETLインスタンスを作成して使用できます。
購入して起動 をクリックして ETL タスクを開始します。