Realtime Compute for Apache Flink:Getting started with batch processing of Realtime Compute for Apache Flink

Step 2: Create tables at the DWD and DWS layers

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

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

Step 3: Create drafts at the DWD and DWS layers and deploy the drafts as deployments

1. Create a draft at the DWD layer and deploy the draft as a deployment.

   1. Create an update draft for a table at the DWD layer.

      On the Development > ETL page, create a blank batch draft named dwd_orders and copy the following SQL statement to the script editor. The INSERT OVERWRITE statement is used to overwrite the table at the DWD layer because the table at the DWD layer is an Apache Paimon append-only 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 SQL Editor page, click Deploy. Then, click OK to deploy the dwd_orders draft as a deployment.

2. Create drafts at the DWS layer and deploy the drafts as deployments.

   1. Create update drafts for tables at the DWS layer.

      On the Development > ETL page, create two blank batch drafts named dws_shops and dws_users. Then, copy one of the following SQL statements to the script editor of the related draft:

      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 SQL Editor page, click Deploy. Then, click OK to deploy the dws_shops and dws_users drafts as deployments.

Step 4: Start and view the deployments at the DWD and DWS layers

• Start and view the deployment at the DWD layer.

  1. On the O&M > Deployments page, select BATCH from the drop-down list of the deployment type. Find the dwd_orders deployment and click Start in the Actions column.

     A batch job in the STARTING state is generated, as shown in the following figure.

     

     If the state of the job changes to FINISHED, data processing is complete.

  2. View the data profiling result.

     In the script editor of the Scripts tab, enter and select the following SQL statement and click Run on the left side of the script editor to query data in the table at the DWD layer:

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

     The following figure shows the query result.

     

• Start and view the deployments at the DWS layer.

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

  2. In the script editor of the Scripts tab, enter and select the following SQL statements and click Run on the left side of the script editor to query data in the tables at the DWS layer:

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

     The following figure shows the query result.

     

Step 5: Orchestrate the deployments as a batch processing workflow

This section describes how to orchestrate the deployments that you created in the previous sections as a workflow. This way, you can run the deployments based on a specific sequence in a unified manner.

1. Create a workflow.

   1. In the left-side navigation pane of the Realtime Compute for Apache Flink console, click O&M > Workflows. On the page that appears, click Create Workflow.

   2. In the Create Workflow panel, enter wf_orders in the Name field, keep the value of the Scheduling Type parameter unchanged (default value: Manual Scheduling), select default-queue for the Resource Queue parameter, and then click Create. The workflow editing page appears.

   3. Edit the workflow.

      1. Click the initial task, name the task v_dwd_orders, and then select the dwd_orders deployment for the task.

      2. Click Add Task to create a task named v_dws_shops. Select the dws_shops deployment for the task and configure the v_dwd_orders task as the upstream task.

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

      4. On the workflow editing page, click Save in the upper-right corner and then click OK.

         The following figure shows the workflow that you created.

         

2. Manually run the workflow.

   Note

   You can change the scheduling type of a workflow to Periodic Scheduling by clicking Edit Workflow in the Actions column of the workflow on the Workflows page. For more information, see Workflows (public preview).

   1. Before you run the workflow, insert some data into the tables at the ODS layer to verify the execution result of the workflow.

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

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

      On the left side of the Results tab, click Flink UI to view the deployment status.

   2. On the O&M > Workflows page, find the workflow that you created and click Execute in the Actions column. In the dialog box that appears, click OK to run the workflow.

      

      Click the name of the workflow to go to the workflow details page. On the Overview tab, view the list of task instances.

      

      Click the ID of a running instance in the workflow to go to the execution details page of the instance. On the page that appears, view the execution status of each task. Wait until the execution of the workflow is complete.

      

3. View the execution result of the workflow.

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

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

   2. View the execution result of the workflow.

      The new data at the ODS layer is processed and written to the tables at the DWS layer.