Realtime Compute for Apache Flink:Flink バッチ処理

ステップ 2: DWD および DWS テーブルの作成

[スクリプト] タブのエディターに、次の SQL ステートメントを入力し、左側にある [実行] をクリックします。

USE `my_catalog`.`order_dw`;

CREATE TABLE dwd_orders (
    order_id BIGINT,
    order_user_id STRING,
    order_shop_id BIGINT,
    order_product_id BIGINT,
    order_product_catalog_name STRING,
    order_fee BIGINT,
    order_create_time TIMESTAMP,
    order_update_time TIMESTAMP,
    order_state INT,
    pay_id BIGINT,
    pay_platform INT COMMENT 'platform 0: phone, 1: pc',
    pay_create_time TIMESTAMP
) WITH (
    'sink.parallelism' = '2'
);

CREATE TABLE dws_users (
    user_id STRING,
    ds STRING,
    total_fee BIGINT COMMENT 'Total amount paid on the current day'
) WITH (
    'sink.parallelism' = '2'
);

CREATE TABLE dws_shops (
    shop_id BIGINT,
    ds STRING,
    total_fee BIGINT COMMENT 'Total amount paid on the current day'
) WITH (
    'sink.parallelism' = '2'
);

ステップ 3: DWD および DWS デプロイメントの作成とデプロイ

1. DWD デプロイメントの作成とデプロイ

   1. DWD テーブル更新デプロイメントの作成

      [開発] > [ETL] ページで、dwd_orders という名前の [空のバッチドラフト] を作成します。推奨タグが付いているエンジンバージョンを選択することをお勧めします。デフォルトの Flink SQL ダイアレクトを使用します。次の SQL ステートメントをエディターにコピーします。DWD テーブルは Paimon の追加専用テーブルであるため、INSERT OVERWRITE ステートメントを使用してテーブル全体を上書きします。

      INSERT OVERWRITE my_catalog.order_dw.dwd_orders
      SELECT 
          o.order_id,
          o.user_id,
          o.shop_id,
          o.product_id,
          c.catalog_name,
          o.buy_fee,
          o.create_time,
          o.update_time,
          o.state,
          p.pay_id,
          p.pay_platform,
          p.create_time
      FROM 
          my_catalog.order_dw.orders as o, 
          my_catalog.order_dw.product_catalog as c, 
          my_catalog.order_dw.orders_pay as p
      WHERE o.product_id = c.product_id AND o.order_id = p.order_id

   2. ページの右上隅で [デプロイ] をクリックし、次に [OK] をクリックして dwd_orders デプロイメントをデプロイします。

2. DWS デプロイメントの作成とデプロイ

   1. DWS テーブル更新デプロイメントの作成

      DWD テーブル更新デプロイメントの作成と同じ手順で、dws_shops と dws_users という名前の 2 つのバッチ下書きを作成します。次の SQL ステートメントをそれぞれのエディターにコピーします。

      INSERT OVERWRITE my_catalog.order_dw.dws_shops
      SELECT 
          order_shop_id,
          DATE_FORMAT(pay_create_time, 'yyyyMMdd') as ds,
          SUM(order_fee) as total_fee
      FROM my_catalog.order_dw.dwd_orders
      WHERE pay_id IS NOT NULL AND order_fee IS NOT NULL
      GROUP BY order_shop_id, DATE_FORMAT(pay_create_time, 'yyyyMMdd');

      INSERT OVERWRITE my_catalog.order_dw.dws_users
      SELECT 
          order_user_id,
          DATE_FORMAT(pay_create_time, 'yyyyMMdd') as ds,
          SUM(order_fee) as total_fee
      FROM my_catalog.order_dw.dwd_orders
      WHERE pay_id IS NOT NULL AND order_fee IS NOT NULL
      GROUP BY order_user_id, DATE_FORMAT(pay_create_time, 'yyyyMMdd');

   2. ページの右上隅で、[Deploy] をクリックし、次に [OK] をクリックして dws_shops と dws_users デプロイメントをデプロイします。

ステップ 4: DWD および DWS デプロイメントの開始と表示

• DWD デプロイメントの開始とデータの表示

  1. [O&M] > [デプロイメント] ページに移動します。 ドロップダウンリストから [BATCH] を選択します。 dwd_orders デプロイメントを見つけ、[操作] 列の [開始] をクリックします。

     状態が [STARTING] のジョブインスタンスが、バッチジョブインスタンスのリストに表示されます。

     ジョブインスタンスのステータスが [FINISHED] に変わると、データ処理が完了します。

  2. データ結果の探索

     [スクリプト] タブのエディターで、次の SQL ステートメントを入力し、左側の [実行] をクリックして DWD テーブルのデータをクエリします。

     SELECT * FROM `my_catalog`.`order_dw`.`dwd_orders`;

     結果は次の図のようになります。

     

• DWS デプロイメントの開始とデータの表示

  1. [運用保守] > [デプロイメント] ページで、ドロップダウンリストから [BATCH] を選択します。dws_shops と dws_users デプロイメントを見つけ、それぞれの [操作] 列で [開始] をクリックします。

  2. [スクリプト] タブのエディターで、以下の SQL ステートメントを入力し、左側の [実行] をクリックして DWS テーブルのデータをクエリします。

     SELECT * FROM `my_catalog`.`order_dw`.`dws_shops`;
     SELECT * FROM `my_catalog`.`order_dw`.`dws_users`;

     結果は次の図のようになります。

     

ステップ 5: タスクオーケストレーションによるバッチパイプラインの構築

このステップでは、デプロイメントをワークフローにオーケストレーションします。これにより、デプロイメントをまとめてトリガーし、順番に実行できます。

1. ワークフローの作成

   1. 左側のナビゲーションウィンドウで、[運用保守] > [ワークフロー] を選択し、[ワークフローの作成] をクリックします。

   2. 表示されたパネルで、名前に wf_orders を入力し、スケジューリングタイプはデフォルト ([手動スケジューリング]) のままにし、[リソースキュー]に default-queue を選択して、[作成] をクリックしてワークフローエディターを開きます。

   3. ワークフローの編集

      1. 最初のノードをクリックし、v_dwd_orders という名前を付け、dwd_orders デプロイメントを選択します。

      2. [タスクの追加] をクリックして v_dws_shops という名前のノードを作成します。dws_shops デプロイメントを選択し、v_dwd_orders を上流ノードとして設定します。

      3. 再度 [タスクの追加] をクリックして v_dws_users という名前のノードを作成します。dws_users デプロイメントを選択し、v_dwd_orders を上流ノードとして設定します。

      4. 右上隅で [保存]、次に [OK] をクリックします。

2. ワークフローの手動トリガー

   説明

   ワークフローを定期スケジュールで実行するように変更することもできます。[ワークフロー] ページで、ワークフローの右側にある [ワークフローの編集] をクリックし、スケジューリングモードを [定期スケジューリング] に変更します。詳細については、「タスクオーケストレーション (パブリックプレビュー)」をご参照ください。

   1. ワークフローをトリガーする前に、ODS テーブルに新しいデータを挿入して、ワークフローの実行結果を検証します。

      [スクリプト] タブのエディターで、次の SQL ステートメントを入力し、左側にある [実行] をクリックします。

      USE `my_catalog`.`order_dw`;
      
      INSERT INTO orders VALUES
      (100008, 'user_001', 12346, 1, 10000, TO_TIMESTAMP('2023-02-15 17:40:56'), TO_TIMESTAMP('2023-02-15 18:42:56'), 1),
      (100009, 'user_002', 12347, 2, 20000, TO_TIMESTAMP('2023-02-15 18:40:56'), TO_TIMESTAMP('2023-02-15 18:42:56'), 1),
      (100010, 'user_003', 12348, 3, 30000, TO_TIMESTAMP('2023-02-15 19:40:56'), TO_TIMESTAMP('2023-02-15 18:42:56'), 1);
      
      INSERT INTO orders_pay VALUES
      (2008, 100008, 1, TO_TIMESTAMP('2023-02-15 20:40:56')),
      (2009, 100009, 1, TO_TIMESTAMP('2023-02-15 20:40:56')),
      (2010, 100010, 1, TO_TIMESTAMP('2023-02-15 20:40:56'));

      [結果] タブの左側にある[Flink UI]をクリックして、ジョブのステータスをモニターします。

   2. [O&M] > [ワークフロー] ページで、前の手順で作成したワークフローを見つけ、[操作] 列の [実行] をクリックし、次に [OK] をクリックしてワークフローをトリガーします。

      ワークフロー名をクリックすると、[ワークフローインスタンスリストと詳細] ページに移動し、そこでワークフローインスタンスリストを表示できます。

      実行中のワークフローインスタンスの[ID]をクリックして、その実行詳細ページに移動し、各ノードのステータスを監視します。ワークフロー全体が完了するまで待ちます。

3. ワークフローの実行結果の表示

   1. [スクリプト] タブのエディターで、次の SQL ステートメントを入力し、左側にある [実行] をクリックします。

      SELECT * FROM `my_catalog`.`order_dw`.`dws_shops`;
      SELECT * FROM `my_catalog`.`order_dw`.`dws_users`;

   2. ワークフローの実行結果の表示

      ワークフローが ODS レイヤーからの新しいデータを処理し、DWS テーブルに書き込んだことがわかります。