Realtime Compute for Apache Flink:Processamento em lote do Flink

Etapa 2: Criar tabelas DWD e DWS

No editor da aba Scripts, insira as seguintes instruções SQL e clique em Run à esquerda.

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

Etapa 3: Criar e implantar jobs DWD e DWS

1. Crie e implante a implantação DWD.

   1. Crie a implantação de atualização da tabela DWD.

      Na página Development > ETL, crie um blank batch draft chamado dwd_orders. Recomendamos selecionar uma versão de engine com a tag recomendada. Use o dialeto Flink SQL padrão. Copie a seguinte instrução SQL para o editor. Como a tabela DWD é do tipo append-only do Paimon, utilize uma instrução INSERT OVERWRITE para sobrescrever toda a tabela.

      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. No canto superior direito da página, clique em Deploy e, em seguida, clique em OK para implantar o job dwd_orders.

2. Crie e implante as implantações DWS.

   1. Crie as implantações de atualização das tabelas DWS.

      Seguindo os mesmos passos descritos em Criar a implantação de atualização da tabela DWD, crie dois rascunhos em lote chamados dws_shops e dws_users. Copie as seguintes instruções SQL para seus 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. No canto superior direito da página, clique em Deploy e, em seguida, clique em OK para implantar os jobs dws_shops e dws_users.

Etapa 4: Iniciar e visualizar implantações DWD e DWS

• Inicie a implantação DWD e visualize seus dados.

  1. Acesse a página O&M > Deployments. Selecione BATCH na lista suspensa. Localize a implantação dwd_orders e clique em Start na coluna Actions.

     Uma instância de job com estado STARTING aparecerá na lista de instâncias de job em lote.

     Quando o status da instância mudar para FINISHED, o processamento de dados estará concluído.

  2. Explore os dados resultantes.

     No editor da aba Scripts, insira a seguinte instrução SQL e clique em Run à esquerda para consultar os dados na tabela DWD.

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

     O resultado é mostrado na figura a seguir.

     

• Inicie as implantações DWS e visualize seus dados.

  1. Na página O&M > Deployments, selecione BATCH na lista suspensa. Localize as implantações dws_shops e dws_users e clique em Start na coluna Actions de cada uma.

  2. No editor da aba Scripts, insira as seguintes instruções SQL e clique em Run à esquerda para consultar os dados nas tabelas DWS.

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

     Os resultados são mostrados nas figuras a seguir.

     

Etapa 5: Construir um pipeline em lote com orquestração de tarefas

Nesta etapa, você orquestrará as implantações em um fluxo de trabalho, permitindo que sejam acionadas juntas e executadas em sequência.

1. Crie um fluxo de trabalho.

   1. No painel de navegação à esquerda, escolha O&M > Workflows e clique em Create Workflow.

   2. No painel exibido, insira wf_orders como nome. Mantenha o tipo de agendamento como padrão (Manual Scheduling), selecione default-queue para a Resource Queue e clique em Create para abrir o editor de fluxo de trabalho.

   3. Edite o fluxo de trabalho.

      1. Clique no nó inicial, nomeie-o como v_dwd_orders e selecione a implantação dwd_orders para ele.

      2. Clique em Add Task para criar um nó chamado v_dws_shops. Selecione a implantação dws_shops e defina v_dwd_orders como nó upstream.

      3. Clique novamente em Add Task para criar um nó chamado v_dws_users. Selecione a implantação dws_users e defina v_dwd_orders como nó upstream.

      4. No canto superior direito, clique em Save e depois em OK.

3. Acione o fluxo de trabalho manualmente.

   Nota

   Também é possível configurar o fluxo para rodar periodicamente. Na página Workflows, clique em Edit Workflow à direita do fluxo desejado e altere o modo de agendamento para Periodic Scheduling. Para mais detalhes, consulte Orquestração de tarefas (Public Preview).

   1. Antes de acionar o fluxo, insira novos dados nas tabelas ODS para validar os resultados da execução.

      No editor da aba Scripts, insira as seguintes instruções SQL e clique em Run à esquerda.

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

      Clique em Flink UI à esquerda da aba Results para monitorar o status do job.

   2. Na página O&M > Workflows, localize o fluxo criado na etapa anterior, clique em Execute na coluna Actions e confirme clicando em OK para iniciar o fluxo.

      Clique no nome do fluxo para acessar a página Workflow Instance List and Details, onde é possível ver a lista de instâncias do fluxo de trabalho.

      Clique no ID da instância em execução para abrir a página de detalhes e acompanhar o status de cada nó. Aguarde a conclusão de todo o fluxo.

4. Visualize os resultados da execução do fluxo.

   1. No editor da aba Scripts, insira as seguintes instruções SQL e clique em Run à esquerda.

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

   2. Verifique os resultados finais.

      Observe que o fluxo processou os novos dados da camada ODS e os gravou nas tabelas DWS.

      

Documentos relacionados

• Para entender melhor os princípios e o ajuste de configuração do processamento em lote do Flink, consulte o Guia de ajuste do processamento em lote do Flink.

• Para construir um data lakehouse de streaming usando Flink e Paimon, consulte Construir um data lakehouse de streaming com Paimon e StarRocks.

• Além de desenvolver jobs Flink no console do Realtime Compute for Apache Flink, você também pode desenvolvê-los localmente. Para mais informações, consulte Desenvolver com a extensão do VS Code.