Realtime Compute for Apache Flink:Flink batch processing

Step 2: Create DWD and DWS tables

In the editor on the Scripts tab, enter the following SQL statements and click Run on the left.

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'
);

Step 3: Create and deploy DWD and DWS deployments

1. Create and deploy the DWD deployment.

   1. Create the DWD table update deployment.

      On the Development > ETL page, create a blank batch draft named dwd_orders. We recommend selecting an engine version with the recommended tag. Use the default Flink SQL dialect. Copy the following SQL statement into the editor. Because the DWD table is a Paimon append-only table, use an INSERT OVERWRITE statement to overwrite the entire table.

      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. In the upper-right corner of the page, click Deploy, and then click OK to deploy the dwd_orders deployment.

2. Create and deploy the DWS deployments.

   1. Create the DWS table update deployments.

      Following the same steps as for creating the DWD table update deployment, create two batch drafts named dws_shops and dws_users. Copy the following SQL statements into their respective editors.

      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. In the upper-right corner of the page, click Deploy, and then click OK to deploy the dws_shops and dws_users deployments.

Step 4: Start and view DWD and DWS deployments

• Start the DWD deployment and view its data.

  1. Go to the O&M > Deployments page. Select BATCH from the drop-down list. Find the dwd_orders deployment and click Start in the Actions column.

     A job instance with the STARTING state appears in the list of batch job instances.

     When the status of the job instance changes to FINISHED, the data processing is complete.

  2. Explore the data results.

     In the editor on the Scripts tab, enter the following SQL statement and click Run on the left to query the data in the DWD table.

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

     The result is shown in the following figure.

     

• Start the DWS deployments and view their data.

  1. On the O&M > Deployments page, select BATCH from the drop-down list. Find the dws_shops and dws_users deployments and click Start in the Actions column for each.

  2. In the editor on the Scripts tab, enter the following SQL statements and click Run on the left to query the data in the DWS tables.

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

     The results are shown in the following figures.

     

Step 5: Build a batch pipeline with task orchestration

In this step, you will orchestrate the deployments into a workflow, which allows them to be triggered together and run in sequence.

1. Create a workflow.

   1. In the left-side navigation pane, choose O&M > Workflows, and then click Create Workflow.

   2. In the panel that appears, enter wf_orders for the name. Keep the scheduling type as the default (Manual Scheduling), select default-queue for the Resource Queue, and then click Create to open the workflow editor.

   3. Edit the workflow.

      1. Click the initial node, name it v_dwd_orders, and select the dwd_orders deployment for it.

      2. Click Add Task to create a node named v_dws_shops. Select the dws_shops deployment for it and set v_dwd_orders as the upstream node.

      3. Click Add Task again to create a node named v_dws_users. Select the dws_users deployment for it and set v_dwd_orders as the upstream node.

      4. In the upper-right corner, click Save and then OK.

2. Manually trigger the workflow.

   Note

   You can also change the workflow to run on a periodic schedule. On the Workflows page, click Edit Workflow to the right of the workflow and change the scheduling mode to Periodic Scheduling. For more information, see Task orchestration (Public Preview).

   1. Before triggering the workflow, insert new data into the ODS tables to verify the workflow's execution results.

      In the editor on the Scripts tab, enter the following SQL statements and click Run on the left.

      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'));

      Click Flink UI on the left of the Results tab to monitor the job status.

   2. On the O&M > Workflows page, find the workflow created in the previous step, click Execute in the Actions column, and then click OK to trigger the workflow.

      Click the workflow name to go to the Workflow Instance List and Details page, where you can view the workflow instance list.

      Click the ID of the running workflow instance to go to its execution details page and observe the status of each node. Wait for the entire workflow to complete.

3. View the workflow execution results.

   1. In the editor on the Scripts tab, enter the following SQL statements and click Run on the left.

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

   2. View the execution results of the workflow.

      You can see that the workflow has processed the new data from the ODS layer and written it into the DWS tables.