Realtime Compute for Apache Flink：Flink批處理快速入門

步驟二：建立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 '當日完成支付的總金額'
) WITH (
    'sink.parallelism' = '2'
);

CREATE TABLE dws_shops (
    shop_id BIGINT,
    ds STRING,
    total_fee BIGINT COMMENT '當日完成支付總金額'
) WITH (
    'sink.parallelism' = '2'
);

步驟三：建立與部署DWD和DWS作業

1. 建立和部署DWD作業。

   1. 建立DWD表更新作業。

      在資料開發 > ETL頁面建立空白的批作業草稿，命名為dwd_orders，將如下SQL語句複製到文本編輯地區中。由於DWD表是Paimon Append Only表，因此此處使用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. 單擊頁面右上方的部署，單擊確定，部署dwd_orders作業。

2. 建立和部署DWS作業。

   1. 建立DWS表更新作業。

      在資料開發 > ETL頁面建立兩個空白的批作業草稿，分別命名為dws_shops和dws_users，將下列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. 單擊頁面右上方的部署，單擊確定，部署dws_shops和dws_users作業。

步驟四：啟動與查看DWD和DWS作業

• 啟動與查看DWD作業資料。

  1. 在營運中心 > 作業營運頁面，在下拉框中選擇批作業，單擊dwd_orders作業操作列下的啟動。

     對應批工作執行個體列表中，產生了一個啟動中的批工作執行個體，如下圖所示。

     

     當該工作執行個體的狀態變為已完成時，表示資料處理完畢。

  2. 探查資料結果。

     在查詢指令碼文本編輯地區，輸入如下SQL語句並單擊左側的運行，查詢DWD表的資料。

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

     結果如下所示。

     

• 啟動與查看DWS作業資料。

  1. 在營運中心 > 作業營運頁面，在下拉框中選擇批作業，單擊dws_shops和dws_users作業操作列下的啟動。

  2. 在查詢指令碼文本編輯地區，輸入如下SQL語句並單擊左側的運行，查詢DWS表的資料。

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

     結果如下所示。

     

步驟五：通過作業編排構建批處理鏈路

本部分將把前面建立的作業編排成一個工作流程，使得它們可以被統一的觸發並有序的執行。

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. 單擊右上方的儲存並確定。

         建立的工作流程如下所示。

         

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. 在營運中心 > 任務編排頁面，單擊上一部分建立的工作流程操作列下的觸發運行，單擊確定，觸發工作流程運行。

      

      單擊工作流程名稱，進入工作流程執行個體列表與詳情頁面，可以看到工作流程執行個體列表。

      

      單擊運行中的工作流程執行個體運行ID，即可進入工作流程執行個體的執行詳情頁面，觀察到各個節點的執行狀態。等待整個工作流程執行完成。

      

3. 查看工作流程執行結果

   1. 在查詢指令碼文本編輯地區，輸入如下SQL語句並單擊左側的運行。

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

   2. 查看工作流程的執行結果。

      可以看到，ODS層新增資料經過處理已經寫入DWS表中。