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表中。