Data Transmission Service:ETL を使用したリアルタイム注文の分析

ユースケース

企業のリアルタイムデータ処理要件を満たすために、ETL 機能はストリームデータの抽出、変換、ロード機能を提供します。大量のリアルタイムデータを効率的に統合し、ドラッグアンドドロップ操作をサポートし、ローコード開発アプローチを使用します。これにより、企業はビジネス上の意思決定を分析し、レポート作成を高速化し、リアルタイムデータを計算できます。デジタルトランスフォーメーションにおいて、リアルタイムデータ処理の一般的なユースケースは次のとおりです：

• 複数リージョンまたは異種データのリアルタイム統合：複数リージョンまたは異種ソースからのデータをリアルタイムで単一のデータベースに保存し、一元化された効率的な管理と意思決定支援を促進します。

• レポート作成の高速化：リアルタイムのレポート作成システムを構築することで、レポート生成効率が大幅に向上し、より多くのリアルタイム分析シナリオをサポートし、デジタルトランスフォーメーションにおけるレポート作成効率に対する高い要求に応えます。

• リアルタイムコンピューティング：ストリーミングビジネスデータをリアルタイムでクレンジングおよび処理して、特徴値とタグを生成します。これにより、ユーザープロファイリング、リスク管理、レコメンデーションシステムなどのオンラインビジネスモデルや、リアルタイムダッシュボード、その他のストリームコンピューティングシナリオをサポートします。

背景情報

この例では、ストリーミング ETL 機能を使用して、リアルタイムのトランザクションデータ (注文 ID、顧客 ID、製品コード、取引額、取引時間など) をビジネスディメンションデータ (製品コード、単価、製品名など) と結合する方法を示します。取引額が 3,000 を超えるなどの特定のフィルター条件を満たすデータは、リアルタイムでデータウェアハウスにロードされます。これにより、製品や顧客などのディメンション別にトランザクションデータの多次元分析が可能になります。また、可視化ツールを使用して、データに基づいたインサイトを得るための動的なダッシュボードを作成することもできます。

操作手順

事前準備

ETL タスクを設定する前に、リアルタイムのトランザクションデータをストリームテーブルに、ビジネスディメンションデータをディメンションテーブルに、ソースの ApsaraDB RDS for MySQL データベースに保存します。

次に、ビジネス要件に基づいて、宛先の ApsaraDB RDS for MySQL データベースにテーブルを作成します。

create table test_orders(
   order_id bigint not null COMMENT '注文 ID',
     user_id bigint not null COMMENT 'ユーザー ID',
     product_id bigint not null COMMENT '製品 ID',
   total_price decimal(15,2) not null COMMENT '注文総額',
   order_date TIMESTAMP  not null COMMENT '注文日',
   PRIMARY KEY (order_id))

CREATE table product (
      product_id bigint not null COMMENT '製品 ID',
      product_name varchar(20) COMMENT '製品名',
      product_price decimal(15,2) not null COMMENT '製品単価')

create table test_orders_new(
 order_id bigint not null COMMENT '注文 ID',
 user_id bigint not null COMMENT 'ユーザー ID',
 product_id bigint not null COMMENT '製品 ID',
 total_price decimal(15,2) not null COMMENT '注文総額',
 order_date TIMESTAMP  not null
 COMMENT '注文日',
 product_id_2 bigint not null COMMENT '製品 ID',
 product_name varchar(20) COMMENT '製品名',
 product_price decimal(15,2) not null COMMENT '製品単価',
 PRIMARY KEY (order_id))

ステップ 1：ソースデータベースの設定

1. ETL タスクリストページに移動します。

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

   2. 左側のナビゲーションペインで、ETL をクリックします。

2. 左上隅で、 をクリックします。データフローの作成 ダイアログボックスで、データフロー名 フィールドに ETL タスクの名前を入力し、開発方法 を DAG に設定します。

3. 確認 をクリックします。

4. ストリームテーブルとディメンションテーブルを設定します。

   1. ストリームテーブルの設定

      1. 左側のナビゲーションペインから、入力/ディメンションテーブル (MySQL) ノードを右側のキャンバスにドラッグします。

      2. キャンバス上の [入力/ディメンションテーブル MySQL-1] ノードをクリックし、ソースデータベースを設定します。

      3. [ノード設定] タブで、次のパラメーターを設定します。

         パラメーター	説明
         [データソース名]	識別しやすいように、説明的な名前を付けることを推奨します。この名前は一意である必要はありません。
         [リージョン]	ソースデータベースのリージョンを選択します。 説明 ETL タスクは、次のリージョンでのみ利用可能です：中国 (杭州)、中国 (上海)、中国 (青島)、中国 (北京)、中国 (張家口)、中国 (深セン)、中国 (広州)、および中国 (香港)。
         [インスタンス]	ソースデータベースをホストするインスタンスの名前を選択します。 インスタンスの作成 をクリックして新しいインスタンスを作成することもできます。 詳細については、「DMS がサポートするデータベース」をご参照ください。
         [ノードタイプ]	ソーステーブルのタイプを選択します。 この例では、ストリームテーブル を選択します。 [ストリームテーブル]：リアルタイムで変更されるテーブルです。 ディメンションテーブルと結合してデータをエンリッチ化できます。 [ディメンションテーブル]：更新頻度が低い (リアルタイムではない) テーブルです。 通常、リアルタイムデータと結合して分析用のワイドテーブルを作成するために使用されます。
         [Convert Format]	データの処理中、ETL はストリームを動的テーブルに変換し、継続的なクエリを実行します。 動的テーブルは、INSERT、UPDATE、DELETE 操作によって常に変更され、新しい動的テーブルが生成されます。 宛先データベースにデータを書き込むとき、この新しい動的テーブルはストリームに変換し直されます。 変更をエンコードするには、[変換フォーマット] を指定する必要があります。 [Upsert ストリーム]：動的テーブル内のデータは、INSERT、UPDATE、DELETE 操作によって変更できます。 ストリームに変換されると、INSERT および UPDATE 操作はアップサートメッセージとしてエンコードされ、DELETE 操作は削除メッセージとしてエンコードされます。 説明 このエンコード形式では、動的テーブルに一意のキー (複合キーも可) が必要です。 [追加専用のストリーム]：動的テーブル内のデータは、INSERT 操作によってのみ変更できます。 ストリームに変換されると、挿入されたデータのみが送信されます。
         [選択したデータベースとテーブル]	変換するソースデータベースとテーブルを選択します。

      4. ノードが設定されると、出力フィールド タブが表示されます。 列名 列で、必要なフィールドを選択します。

      5. この例では、ストリームテーブルが使用されます。 時間属性 タブをクリックし、次のパラメーターを設定します。

         パラメーター	説明
         [イベント時刻のウォーターマークを選択]	ストリームテーブルから時間フィールドを選択します。ストリームテーブルには通常、ordertime のようなビジネス固有のタイムスタンプなど、データが生成された日時を示す時間フィールドが含まれています。
         [イベント時刻ウォーターマークの遅延]	許容できる最大のデータレイテンシーを入力します。 ネットワークの状態により、データが生成された順序で ETL サービスに到着しない場合があり、遅延が発生する可能性があります。 ETL は、遅れて到着するデータを無期限に待機することはできません。 したがって、順序が乱れたデータを処理するには、レイテンシー期間を指定する必要があります。 たとえば、10:00 のデータが到着したが、9:59 のデータが到着していない場合、ETL は 10:00 に指定されたレイテンシーを加えた時間までしか待機しません。 9:59 のデータがこの期間を過ぎて到着した場合、ETL はそのデータを破棄します。
         [処理時間]	処理時間は、ETL がデータを処理するときのローカルサーバー時間です。 この値を格納する列を定義する必要があります。 これは主に演算子の計算に使用されます。たとえば、テンポラル結合では、処理時間を使用して通常テーブルの最新バージョンを検索します。

      説明

      ソースデータベースノードの右側にある感嘆符アイコン が消えたら、設定は完了です。

   2. ディメンションテーブルの設定

      1. 左側のナビゲーションペインから、入力/ディメンションテーブル (MySQL) ノードを右側のキャンバスにドラッグします。

      2. キャンバス上の [入力/ディメンションテーブル MySQL-2] ノードをクリックし、ソースデータベースを設定します。

      3. [ノード設定] タブで、次のパラメーターを設定します。

         パラメーター	説明
         [データソース名]	識別しやすいように、説明的な名前を付けることを推奨します。この名前は一意である必要はありません。
         [リージョン]	ソースデータベースのリージョンを選択します。
         [インスタンス]	ソースデータベースをホストするインスタンスの名前を選択します。 インスタンスの作成 をクリックして新しいインスタンスを作成することもできます。 詳細については、「DMS がサポートするデータベース」をご参照ください。
         [ノードタイプ]	ソーステーブルのタイプを選択します。 この例では、ディメンションテーブル を選択します。
         [選択したデータベースとテーブル]	変換するソースデータベースとテーブルを選択します。

      4. ノードが設定されると、出力フィールド タブが表示されます。 列名 列で、必要なフィールドを選択します。

      説明

      ソースデータベースノードの右側にある感嘆符アイコン が消えたら、設定は完了です。

ステップ 2：テーブル結合ノードの設定

1. 左側のペインで、テーブル結合 ノードをキャンバス上の空白領域にドラッグします。

2. 設定済みのストリームテーブルノードとディメンションテーブルノードの上にポインターを移動します。 各ノードの円をクリックし、接続線を [テーブル結合-1] ノードにドラッグします。

3. キャンバス上の [テーブル結合-1] ノードをクリックして、ノードを設定します。

   1. ノード設定 タブで、次のパラメーターを設定します。

      セクション	パラメーター	説明
      [変換名]	[変換名の入力]	識別しやすいよう、説明的な名前を付けることを推奨します。この名前は一意である必要はありません。
      [JOIN 設定]	[JOIN 句の左のテーブル]	JOIN 演算子の左側に配置するテーブルを選択します。 このテーブルがプライマリテーブルとして機能します。 この例では、ストリームテーブルを選択します。
      	[一時結合の時間属性 (選択されていない場合、通常の結合が適用されます)]	テンポラルテーブルと結合するために使用されるストリームテーブルの時間属性を選択します。 属性を選択しない場合、DTS は通常の結合を実行します。 この例では、処理時間に基づく を選択します。 説明 テンポラルテーブル (動的テーブルとも呼ばれる) は、時間の経過に伴うデータの変更を記録するテーブルのパラメーター化されたビューです。 テンポラルテーブルには、データの履歴バージョンを示すバージョン管理されたテーブルと、データの最新バージョンのみを示す通常テーブルが含まれます。 テンポラル結合では、ストリームテーブルに時間属性があり、右テーブルにプライマリキーが必要です。 右テーブルがディメンションテーブルの場合、[JOIN 条件] にはディメンションテーブルのプライマリキーを含める必要があります。 [イベント時刻ウォーターマークに基づきます]：ストリームテーブルのイベントタイムを使用して、バージョン管理されたテーブルの対応するバージョンと結合します。 [処理時間に基づく]：ストリームテーブルの処理時間を使用して、通常テーブルの最新バージョンと結合します。
      	[JOIN 演算の選択]	結合操作を選択します。 この例では、[内部結合] を選択します。 内部結合：両方のテーブルに一致する値を持つレコードを返します。 左結合：左テーブルのすべてのレコードと、右テーブルの一致したレコードを返します。 右結合：右テーブルのすべてのレコードと、左テーブルの一致したレコードを返します。
      [JOIN 条件]	[+ 条件の追加]	[+ 条件の追加] をクリックして、結合条件のフィールドを選択します。 説明 等号 (=) の左側のフィールドは結合の左テーブルに属し、右側のフィールドは右テーブルに属します。

4. 結合条件を設定した後、出力フィールド タブをクリックします。 列名 列で、必要なフィールドを選択します。

ステップ 3：テーブルレコードフィルターノードの設定

1. 左側のペインで、テーブルレコードフィルター ノードをキャンバス上の空白領域にドラッグします。

2. 設定済みの [テーブル結合-1] ノードの上にポインターを移動します。 円をクリックし、[テーブル結合-1] ノードから [テーブルレコードフィルター-1] ノードに接続線をドラッグします。

3. キャンバス上の [テーブルレコードフィルター-1] ノードをクリックして、ノードを設定します。

   1. 変換名 セクションで、変換の名前を入力します。

      説明

      識別しやすいよう、説明的な名前を付けることを推奨します。この名前は一意である必要はありません。

   2. WHERE 条件 セクションで、次のいずれかの方法を使用して WHERE 条件を設定します。

      • WHERE 条件を直接入力します。たとえば、total_price > 3000.00 と入力すると、結合テーブルの total_price が 3000.00 より大きいレコードが絞り込まれます。

      • 入力フィールド または オペレーター セクションのオプションをクリックして、WHERE 条件を構築します。

ステップ 4：宛先データベースの設定

1. ページの左側にあるナビゲーションペインで、[出力 MySQL] ノードをページの右側にあるキャンバスの空白領域にドラッグします。

2. 設定済みの [テーブルレコードフィルター-1] ノードの上にポインターを移動します。 円をクリックし、[テーブルレコードフィルター-1] ノードから [出力 MySQL-1] ノードに接続線をドラッグします。

3. キャンバス上の [出力 MySQL-1] ノードをクリックして、宛先データベースを設定します。

   1. [ノード設定] タブで、次のパラメーターを設定します。

      パラメーター	説明
      [データソース名]	識別しやすいように、説明的な名前を付けることを推奨します。この名前は一意である必要はありません。
      [リージョン]	宛先データベースのリージョンを選択します。 説明 ETL タスクは、次のリージョンでのみ利用可能です：中国 (杭州)、中国 (上海)、中国 (青島)、中国 (北京)、中国 (張家口)、中国 (深セン)、中国 (広州)、および中国 (香港)。
      [インスタンス]	宛先データベースをホストするインスタンスの名前を選択します。 インスタンスの作成 をクリックして新しいインスタンスを作成することもできます。 詳細については、「DMS がサポートするデータベース」をご参照ください。
      [テーブルマッピング]	変換されたデータが保存される宛先テーブルを選択します。 移行先テーブルの選択 セクションで、宛先テーブルをクリックします。

4. 列名 列で、必要なパラメーターを選択します。

ステップ 5：事前チェックとタスクの開始

1. 設定が完了したら、Flink SQL 検証の生成 をクリックします。 ETL は Flink SQL を生成して検証します。

2. 検証が完了したら、ETL 検証の詳細を表示する をクリックして、生成された Flink SQL ステートメントと検証結果をダイアログボックスで表示します。 情報が正しいことを確認したら、閉じる をクリックします。

   説明

   検証に失敗した場合は、結果に表示された原因に基づいて障害をトラブルシューティングします。

3. 次：タスク設定の保存と事前チェック をクリックします。 ETL タスクは、事前チェックに合格した後にのみ開始できます。 事前チェックに失敗した場合は、失敗した項目の横にある 詳細を表示 をクリックし、表示された情報に基づいて問題を修正し、事前チェックを再度実行します。

4. 事前チェックが完了したら、ページの下部にある 次：インスタンスの購入 をクリックします。

5. 購入 ページで、インスタンスクラス を選択し、[コンピューティングユニット (CU)] を指定します。 次に、[Data Transmission Service (従量課金) サービス規約] と [パブリックプレビューのサービス規約] を読んで選択します。

   説明

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

6. 購入して起動 をクリックして ETL タスクを開始します。

タスク結果

この例では、ETL タスクが開始された後 (例えば 8 月 1 日に)、test_orders リアルタイムトランザクションテーブルで更新されたレコードがフィルター条件 (total_price > 3000.00) を満たすと、そのレコードは test_orders_new 宛先テーブルに同期されます。