Realtime Compute for Apache Flink:Traitement par lots Flink

Étape 2 : Créer des tables DWD et DWS

Dans l'éditeur de l'onglet Scripts, saisissez les instructions SQL suivantes et cliquez sur Run à gauche.

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

Étape 3 : Créer et déployer des déploiements DWD et DWS

1. Créez et déployez le déploiement DWD.

   1. Créez le déploiement de mise à jour de la table DWD.

      Sur la page Development > ETL, créez un blank batch draft nommé dwd_orders. Nous vous recommandons de sélectionner une version du moteur portant le tag recommandé. Utilisez le dialecte Flink SQL par défaut. Copiez l'instruction SQL suivante dans l'éditeur. Étant donné que la table DWD est une table Paimon en ajout seul, utilisez une instruction INSERT OVERWRITE pour écraser l'intégralité de la 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. Dans le coin supérieur droit de la page, cliquez sur Deploy, puis sur OK pour déployer le déploiement dwd_orders.

2. Créez et déployez les déploiements DWS.

   1. Créez les déploiements de mise à jour des tables DWS.

      En suivant les mêmes étapes que pour la création du déploiement de mise à jour de la table DWD, créez deux brouillons par lots nommés dws_shops et dws_users. Copiez les instructions SQL suivantes dans leurs éditeurs respectifs.

      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. Dans le coin supérieur droit de la page, cliquez sur Deploy, puis sur OK pour déployer les déploiements dws_shops et dws_users.

Étape 4 : Démarrer et afficher les déploiements DWD et DWS

• Démarrez le déploiement DWD et affichez ses données.

  1. Accédez à la page O&M > Deployments. Sélectionnez BATCH dans la liste déroulante. Recherchez le déploiement dwd_orders et cliquez sur Start dans la colonne Actions.

     Une instance de job dont l'état est STARTING apparaît dans la liste des instances de jobs par lots.

     Lorsque l'état de l'instance de job passe à FINISHED, le traitement des données est terminé.

  2. Explorez les résultats des données.

     Dans l'éditeur de l'onglet Scripts, saisissez l'instruction SQL suivante et cliquez sur Run à gauche pour interroger les données de la table DWD.

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

     Le résultat est illustré dans la figure suivante.

     

• Démarrez les déploiements DWS et affichez leurs données.

  1. Sur la page O&M > Deployments, sélectionnez BATCH dans la liste déroulante. Recherchez les déploiements dws_shops et dws_users et cliquez sur Start dans la colonne Actions pour chacun d'eux.

  2. Dans l'éditeur de l'onglet Scripts, saisissez les instructions SQL suivantes et cliquez sur Run à gauche pour interroger les données des tables DWS.

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

     Les résultats sont illustrés dans les figures suivantes.

     

Étape 5 : Construire un pipeline par lots avec l'orchestration des tâches

À cette étape, vous orchestrez les déploiements au sein d'un workflow, ce qui permet de les déclencher ensemble et de les exécuter séquentiellement.

1. Créez un workflow.

   1. Dans le volet de navigation de gauche, choisissez O&M > Workflows, puis cliquez sur Create Workflow.

   2. Dans le panneau qui s'affiche, saisissez wf_orders comme nom. Conservez le type de planification par défaut (Manual Scheduling), sélectionnez default-queue pour la Resource Queue, puis cliquez sur Create pour ouvrir l'éditeur de workflow.

   3. Modifiez le workflow.

      1. Cliquez sur le nœud initial, nommez-le v_dwd_orders et sélectionnez le déploiement dwd_orders pour celui-ci.

      2. Cliquez sur Add Task pour créer un nœud nommé v_dws_shops. Sélectionnez le déploiement dws_shops pour celui-ci et définissez v_dwd_orders comme nœud amont.

      3. Cliquez à nouveau sur Add Task pour créer un nœud nommé v_dws_users. Sélectionnez le déploiement dws_users pour celui-ci et définissez v_dwd_orders comme nœud amont.

      4. Dans le coin supérieur droit, cliquez sur Save puis sur OK.

2. Déclenchez manuellement le workflow.

   Remarque

   Vous pouvez également configurer le workflow pour qu'il s'exécute selon une planification périodique. Sur la page Workflows, cliquez sur Edit Workflow à droite du workflow et modifiez le mode de planification en Periodic Scheduling. Pour plus d'informations, consultez la rubrique Orchestration des tâches (Aperçu public).

   1. Avant de déclencher le workflow, insérez de nouvelles données dans les tables ODS pour vérifier les résultats d'exécution du workflow.

      Dans l'éditeur de l'onglet Scripts, saisissez les instructions SQL suivantes et cliquez sur Run à gauche.

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

      Cliquez sur Flink UI à gauche de l'onglet Results pour surveiller l'état du job.

   2. Sur la page O&M > Workflows, recherchez le workflow créé à l'étape précédente, cliquez sur Execute dans la colonne Actions, puis sur OK pour déclencher le workflow.

      Cliquez sur le nom du workflow pour accéder à la page Overview, où vous pouvez afficher la liste des instances de workflow.

      Cliquez sur l'ID de l'instance de workflow en cours d'exécution pour accéder à sa page de détails d'exécution et observer l'état de chaque nœud. Attendez que l'ensemble du workflow soit terminé.

3. Affichez les résultats d'exécution du workflow.

   1. Dans l'éditeur de l'onglet Scripts, saisissez les instructions SQL suivantes et cliquez sur Run à gauche.

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

   2. Consultez les résultats d'exécution du workflow.

      Vous constatez que le workflow a traité les nouvelles données de la couche ODS et les a écrites dans les tables DWS.