Realtime Compute for Apache Flink:Procesamiento por lotes de Flink

Paso 2: Crear tablas DWD y DWS

En el editor de la pestaña Scripts, introduzca las siguientes sentencias SQL y haga clic en Run a la izquierda.

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

Paso 3: Crear y desplegar los despliegues DWD y DWS

1. Cree y despliegue el despliegue DWD.

   1. Cree el despliegue de actualización de la tabla DWD.

      En la página Development > ETL, cree un blank batch draft llamado dwd_orders. Se recomienda seleccionar una versión de motor con la etiqueta recomendada. Use el dialecto Flink SQL predeterminado. Copie la siguiente sentencia SQL en el editor. Dado que la tabla DWD es una tabla Paimon de solo anexión, utilice una sentencia INSERT OVERWRITE para sobrescribir toda la tabla.

      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. En la esquina superior derecha de la página, haga clic en Deploy y luego en OK para desplegar el despliegue dwd_orders.

2. Cree y despliegue los despliegues DWS.

   1. Cree los despliegues de actualización de las tablas DWS.

      Siguiendo los mismos pasos que para creating the DWD table update deployment, cree dos borradores por lotes llamados dws_shops y dws_users. Copie las siguientes sentencias SQL en sus respectivos editores.

      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. En la esquina superior derecha de la página, haga clic en Deploy y luego en OK para desplegar los despliegues dws_shops y dws_users.

Paso 4: Iniciar y consultar los despliegues DWD y DWS

• Inicie el despliegue DWD y consulte sus datos.

  1. Vaya a la página O&M > Deployments. Seleccione BATCH en la lista desplegable. Busque el despliegue dwd_orders y haga clic en Start en la columna Actions.

     Aparecerá una instancia de trabajo con estado STARTING en la lista de instancias de trabajo por lotes.

     Cuando el estado de la instancia de trabajo cambie a FINISHED, el procesamiento de datos habrá finalizado.

  2. Explore los resultados de los datos.

     En el editor de la pestaña Scripts, introduzca la siguiente sentencia SQL y haga clic en Run a la izquierda para consultar los datos en la tabla DWD.

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

     El resultado se muestra en la siguiente figura.

     

• Inicie los despliegues DWS y consulte sus datos.

  1. En la página O&M > Deployments, seleccione BATCH en la lista desplegable. Busque los despliegues dws_shops y dws_users, y haga clic en Start en la columna Actions de cada uno.

  2. En el editor de la pestaña Scripts, introduzca las siguientes sentencias SQL y haga clic en Run a la izquierda para consultar los datos en las tablas DWS.

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

     Los resultados se muestran en las siguientes figuras.

     

Paso 5: Construir un pipeline por lotes con orquestación de tareas

En este paso, orquestará los despliegues en un flujo de trabajo, lo que permite activarlos conjuntamente y ejecutarlos en secuencia.

1. Cree un flujo de trabajo.

   1. En el panel de navegación izquierdo, elija O&M > Workflows y luego haga clic en Create Workflow.

   2. En el panel que aparece, introduzca wf_orders como nombre. Mantenga el tipo de programación predeterminado (Manual Scheduling), seleccione default-queue para la Resource Queue y luego haga clic en Create para abrir el editor de flujos de trabajo.

   3. Edite el flujo de trabajo.

      1. Haga clic en el nodo inicial, asígnele el nombre v_dwd_orders y seleccione el despliegue dwd_orders para dicho nodo.

      2. Haga clic en Add Task para crear un nodo llamado v_dws_shops. Seleccione el despliegue dws_shops para este nodo y configure v_dwd_orders como nodo ascendente.

      3. Haga clic nuevamente en Add Task para crear un nodo llamado v_dws_users. Seleccione el despliegue dws_users para este nodo y configure v_dwd_orders como nodo ascendente.

      4. En la esquina superior derecha, haga clic en Save y luego en OK.

2. Active manualmente el flujo de trabajo.

   Nota

   También puede configurar el flujo de trabajo para que se ejecute según una programación periódica. En la página Workflows, haga clic en Edit Workflow a la derecha del flujo de trabajo y cambie el modo de programación a Periodic Scheduling. Para obtener más información, consulte Task orchestration (Public Preview).

   1. Antes de activar el flujo de trabajo, inserte nuevos datos en las tablas ODS para verificar los resultados de ejecución del flujo.

      En el editor de la pestaña Scripts, introduzca las siguientes sentencias SQL y haga clic en Run a la izquierda.

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

      Haga clic en Flink UI a la izquierda de la pestaña Results para supervisar el estado del trabajo.

   2. En la página O&M > Workflows, busque el flujo de trabajo creado en el paso anterior, haga clic en Execute en la columna Actions y luego en OK para activar el flujo de trabajo.

      Haga clic en el nombre del flujo de trabajo para ir a la página Overview, donde puede consultar la lista de instancias del flujo de trabajo.

      Haga clic en el ID de la instancia del flujo de trabajo en ejecución para ir a su página de detalles de ejecución y observar el estado de cada nodo. Espere a que finalice todo el flujo de trabajo.

3. Consulte los resultados de ejecución del flujo de trabajo.

   1. En el editor de la pestaña Scripts, introduzca las siguientes sentencias SQL y haga clic en Run a la izquierda.

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

   2. Revise los resultados de ejecución del flujo de trabajo.

      Podrá observar que el flujo de trabajo ha procesado los nuevos datos de la capa ODS y los ha escrito en las tablas DWS.