Realtime Compute for Apache Flink:Realtime Compute for Apache 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 of payment that is complete on the current day'
) WITH (
    'sink.parallelism' = '2'
);

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

手順 3: DWD レイヤーと DWS レイヤーにドラフトを作成し、ドラフトをデプロイメントとしてデプロイする

1. DWD レイヤーにドラフトを作成し、ドラフトをデプロイメントとしてデプロイします。

   1. DWD レイヤーのテーブルの更新ドラフトを作成します。

      [開発] > [ETL] ページで、dwd_orders という名前の空のバッチドラフトを作成し、次の SQL 文をスクリプトエディターにコピーします。「RECOMMENDED」ラベルが付いた VVR バージョンを選択することをお勧めします。デフォルトでは、Flink SQL 文が使用され、以前のバージョンのエンジンでは SQL ダイアレクトは使用できません。DWD レイヤーのテーブルは Apache Paimon 追加専用テーブルであるため、INSERT OVERWRITE 文を使用して DWD レイヤーのテーブルを上書きします。

      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. SQL エディターページの右上隅にある [デプロイ] をクリックします。次に、[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. SQL エディターページの右上隅にある [デプロイ] をクリックします。次に、[OK] をクリックして、dws_shops ドラフトと dws_users ドラフトをデプロイメントとしてデプロイします。

手順 4: DWD レイヤーと DWS レイヤーのデプロイメントを開始して表示する

• DWD レイヤーのデプロイメントを開始して表示します。

  1. [O&M] > [デプロイメント] ページで、デプロイメントタイプのドロップダウンリストから [バッチ] を選択します。dwd_orders デプロイメントを見つけ、[アクション] 列の [開始] をクリックします。

     次の図に示すように、[開始中] 状態のバッチジョブが生成されます。

     

     ジョブの状態が [完了] に変わると、データ処理は完了です。

  2. データプロファイリング結果を表示します。

     [スクリプト] タブのスクリプトエディターに、次の SQL 文を入力して選択し、スクリプトエディターの左側にある [実行] をクリックして、DWD レイヤーのテーブルのデータをクエリします。

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

     次の図は、クエリ結果を示しています。

     

• DWS レイヤーのデプロイメントを開始して表示します。

  1. [O&M] > [デプロイメント] ページで、デプロイメントタイプのドロップダウンリストから [バッチ] を選択します。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. Realtime Compute for Apache Flink コンソールの左側のナビゲーションウィンドウで、[O&M] > [ワークフロー] をクリックします。表示されるページで、[ワークフローの作成] をクリックします。

   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 レイヤーのテーブルに書き込まれます。