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(
 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. DTS コンソールにログインします。

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

2. 左上隅の アイコンをクリックします。データフローの作成 ダイアログボックスで、データフロー名 フィールドに名前を入力し、開発方法 を [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. 構成済みのストリームテーブルノードとディメンションテーブルノードにポインターを合わせます。各ノードの丸印をクリックし、接続線を [Table Join-1] ノードまでドラッグします。

3. キャンバス上の [Table Join-1] ノードをクリックして、ノードを構成します。

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

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

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

手順 3： Table Record Filter ノードの設定

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

2. 設定済みの [Table Join-1] ノードにポインターを合わせます。 円をクリックし、[Table Join-1] ノードから [Table Record Filter-1] ノードまで接続線をドラッグします。

3. キャンバス上の [Table Record Filter-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)] を指定し、[データ伝送サービス (従量課金) サービス規約] と [パブリックプレビューサービス規約] を読んで選択します。

   説明

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

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

タスク結果

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