Realtime Compute for Apache Flink:Streaming lakehouse com Paimon e StarRocks

Preparar a fonte de dados MySQL CDC****

Este tutorial usa o ApsaraDB RDS for MySQL como exemplo. Você criará um banco de dados chamado order_dw e três tabelas com dados de amostra.

1. {{XREF_7}}.

   Importante

   Certifique-se de que a instância do ApsaraDB RDS for MySQL e seu workspace do Realtime Compute for Apache Flink estejam na mesma VPC. Se estiverem em VPCs diferentes, consulte {{XREF_8}}

2. {{XREF_9}}.

   Crie um banco de dados chamado order_dw. Em seguida, crie uma conta privilegiada ou uma conta padrão com permissões de leitura e gravação para o banco de dados order_dw.

   Crie três tabelas e insira dados nelas.

   SELECT
     order_create_time AS order_create_date,
     order_product_catalog_name,
     COUNT(*),
     SUM(order_fee)
   FROM
     paimoncatalog.order_dw.dwd_orders
   WHERE
     order_create_time >= '2023-02-01 00:00:00'  and order_create_time < '2023-03-01 00:00:00'
   GROUP BY
     order_create_date, order_product_catalog_name
   ORDER BY
     order_create_date, order_product_catalog_name;

Gerenciar metadados

Criar um catálogo Paimon

1. Faça login no console do Realtime Compute for Apache Flink.

2. No painel de navegação à esquerda, acesse Metadata Management e clique em Create Catalog.

3. Na aba Built-in Catalog, clique em Apache Paimon e, em seguida, clique em Next.

4. Configure os parâmetros a seguir, selecione DLF como tipo de armazenamento e clique em OK.

   Parâmetro	Descrição	Obrigatório	Observações
   metastore	O tipo de metastore.	Sim	Neste exemplo, selecione DLF.
   catalog name	O nome do catálogo de dados do DLF. Importante Se você usar um usuário RAM ou uma função RAM, certifique-se de que o usuário RAM ou a função RAM tenha as permissões necessárias para ler e gravar dados no DLF. Para obter mais informações, consulte {{XREF_10}}.	Sim	Recomendamos que você use {{XREF_11}}. Esta versão elimina a necessidade de inserir informações como um par de AccessKey e permite selecionar rapidamente um catálogo de dados DLF existente. Para obter informações sobre como criar um catálogo de dados, consulte {{XREF_12}}. Após criar um catálogo de dados chamado paimoncatalog, selecione-o na lista.

5. No catálogo de dados, crie um banco de dados chamado order_dw para sincronizar todas as tabelas do banco de dados order_dw do MySQL.

   No painel de navegação à esquerda, escolha Data Exploration > Query Editor e clique em New para abrir uma janela de consulta temporária.

   SELECT
     order_create_time AS order_create_date,
     order_product_catalog_name,
     COUNT(*),
     SUM(order_fee)
   FROM
     paimoncatalog.order_dw.dwd_orders
   WHERE
     order_create_time >= '2023-02-01 00:00:00'  and order_create_time < '2023-03-01 00:00:00'
   GROUP BY
     order_create_date, order_product_catalog_name
   ORDER BY
     order_create_date, order_product_catalog_name;

   A mensagem retornada The following statement has been executed successfully! indica que o banco de dados foi criado com sucesso.

Para obter mais informações sobre como usar catálogos Paimon, consulte {{XREF_13}}.

Criar um catálogo MySQL

1. Na página Metadata Management, clique em Create Catalog.

2. Na aba Built-in Catalog, clique em MySQL e, em seguida, clique em Next.

3. Configure os parâmetros a seguir e clique em OK para criar um catálogo MySQL chamado mysqlcatalog.

   Parâmetro	Descrição	Obrigatório	Observações
   catalog name	O nome do catálogo.	Sim	Insira um nome personalizado. Este tutorial usa mysqlcatalog.
   hostname	O endereço IP ou hostname do banco de dados MySQL.	Sim	Para obter mais informações, consulte {{XREF_14}}. Como a instância do ApsaraDB RDS for MySQL e o workspace do Realtime Compute for Apache Flink estão na mesma VPC, insira o endpoint interno.
   port	O número da porta do service de banco de dados MySQL. O valor padrão é 3306.	Não	Para obter mais informações, consulte {{XREF_15}}.
   default-database	O nome do banco de dados MySQL padrão.	Sim	Insira o nome do banco de dados a ser sincronizado, que é order_dw neste tutorial.
   username	O nome de usuário para o service de banco de dados MySQL.	Sim	Use a conta criada em {{XREF_16}}.
   password	A senha para o service de banco de dados MySQL.	Sim	Use a senha da conta criada em {{XREF_17}}.

Construir a camada ODS: Ingestão em tempo real

Construa a camada ODS em uma única etapa usando um job de ingestão de dados Flink CDC no formato YAML para sincronizar dados do MySQL para o Paimon.

1. Crie e inicie o job de ingestão de dados YAML.

   1. No console do Realtime Compute for Apache Flink, acesse a página Data Development > Data Ingestion e crie um job de rascunho YAML em branco chamado ods.

   2. Copie o código a seguir para o editor e atualize parâmetros como nome de usuário e senha.

      SELECT
        order_create_time AS order_create_date,
        order_product_catalog_name,
        COUNT(*),
        SUM(order_fee)
      FROM
        paimoncatalog.order_dw.dwd_orders
      WHERE
        order_create_time >= '2023-02-01 00:00:00'  and order_create_time < '2023-03-01 00:00:00'
      GROUP BY
        order_create_date, order_product_catalog_name
      ORDER BY
        order_create_date, order_product_catalog_name;

      Parâmetro	Descrição	Obrigatório	Exemplo
      catalog.properties.metastore	O tipo de metastore. Defina como rest.	Sim	rest
      catalog.properties.token.provider	O provedor de token. Defina o valor como dlf.	Sim	dlf
      catalog.properties.uri	A URI de acesso para o DLF Rest Catalog Server é http://[region-id]-vpc.dlf.aliyuncs.com. Para obter mais informações sobre o ID da Região, consulte {{XREF_18}}.	Sim	http://cn-beijing-vpc.dlf.aliyuncs.com
      catalog.properties.warehouse	O nome do catálogo DLF.	Sim	paimoncatalog
      hostname	O endereço IP ou hostname do banco de dados MySQL. Para obter mais informações, consulte {{XREF_19}}. Como a instância do ApsaraDB RDS for MySQL e o workspace do Realtime Compute for Apache Flink estão na mesma VPC, insira o endpoint interno.	Sim	rm-bp1e****566g.mysql.rds.aliyuncs.com
      username	O nome de usuário do banco de dados MySQL. Recomendamos o uso de gerenciamento de segredos. Para obter mais informações, consulte {{XREF_20}}.	Sim	${secret_values.username}
      password	A senha do banco de dados MySQL. Recomendamos o uso de gerenciamento de segredos. Para obter mais informações, consulte {{XREF_21}}.	Sim	${secret_values.password}

      Para otimizar o desempenho de gravação do Paimon, consulte {{XREF_22}}.

   3. No canto superior direito, clique em Deploy.

   4. Em O&M Center > Job O&M, clique em Start na coluna Actions do job ODS recém-implantado e selecione stateless start para iniciar o job. Para obter mais informações sobre configurações de inicialização de job, consulte {{XREF_23}}.

2. Visualize os dados nas três tabelas que foram sincronizadas do MySQL para o Paimon.

   No console do Realtime Compute for Apache Flink, acesse a página Data Development > Data Exploration. Na aba Query Editor, copie o código a seguir para o editor, selecione a instrução e clique em Run no canto superior direito.

   SELECT
     order_create_time AS order_create_date,
     order_product_catalog_name,
     COUNT(*),
     SUM(order_fee)
   FROM
     paimoncatalog.order_dw.dwd_orders
   WHERE
     order_create_time >= '2023-02-01 00:00:00'  and order_create_time < '2023-03-01 00:00:00'
   GROUP BY
     order_create_date, order_product_catalog_name
   ORDER BY
     order_create_date, order_product_catalog_name;

   Após a execução da consulta, sete registros de pedidos da tabela orders são retornados. O resultado inclui as colunas order_id (100001 a 100007), user_id, shop_id, product_id, buy_fee (1000 a 5000), create_time, update_time e state. O valor para state é 1 para todos os registros.

Construir a camada DWD: tabela larga temática

1. Crie a tabela larga dwd_orders

   No console do Realtime Compute for Apache Flink, acesse a página Data Development > Data Exploration. Na aba Query Editor, copie o código a seguir para o editor, selecione a instrução e clique em Run no canto superior direito.

   SELECT
     order_create_time AS order_create_date,
     order_product_catalog_name,
     COUNT(*),
     SUM(order_fee)
   FROM
     paimoncatalog.order_dw.dwd_orders
   WHERE
     order_create_time >= '2023-02-01 00:00:00'  and order_create_time < '2023-03-01 00:00:00'
   GROUP BY
     order_create_date, order_product_catalog_name
   ORDER BY
     order_create_date, order_product_catalog_name;

   A mensagem The following statement has been executed successfully! indica uma criação bem-sucedida.

2. Consuma dados de alteração do ODS

   No console do Realtime Compute for Apache Flink, acesse a página Data Development > ETL. Crie um novo job de streaming SQL chamado dwd, copie o código a seguir para o editor SQL, clique em Deploy e inicie o job usando Stateless Start.

   Este job SQL une a tabela orders com a tabela product_catalog em uma junção de tabela de dimensão. Em seguida, ele grava o resultado na tabela dwd_orders junto com os dados da tabela orders_pay. O mecanismo de mesclagem de atualização parcial do Paimon une registros das tabelas orders e orders_pay que possuem o mesmo order_id.

   SELECT
     order_create_time AS order_create_date,
     order_product_catalog_name,
     COUNT(*),
     SUM(order_fee)
   FROM
     paimoncatalog.order_dw.dwd_orders
   WHERE
     order_create_time >= '2023-02-01 00:00:00'  and order_create_time < '2023-03-01 00:00:00'
   GROUP BY
     order_create_date, order_product_catalog_name
   ORDER BY
     order_create_date, order_product_catalog_name;

3. Visualize os dados em dwd_orders

   No console do Realtime Compute for Apache Flink, acesse a página Data Development > Data Exploration. Na aba Query Editor, copie o código a seguir para o editor, selecione a instrução e clique em Run no canto superior direito.

   SELECT
     order_create_time AS order_create_date,
     order_product_catalog_name,
     COUNT(*),
     SUM(order_fee)
   FROM
     paimoncatalog.order_dw.dwd_orders
   WHERE
     order_create_time >= '2023-02-01 00:00:00'  and order_create_time < '2023-03-01 00:00:00'
   GROUP BY
     order_create_date, order_product_catalog_name
   ORDER BY
     order_create_date, order_product_catalog_name;

   Após a execução bem-sucedida da consulta, sete registros de pedidos da tabela larga dwd_orders são retornados. O resultado inclui oito campos: order_id, order_user_id, order_shop_id, order_product_id, order_product_catalog_name, order_fee, order_create_time e order_update_time. O order_id varia de 100001 a 100007, as taxas de pedido variam de 1000 a 5000 e todos os horários de criação são em 2023-02-15.

Construir a camada DWS: cálculo de métricas

1. Crie as tabelas agregadas da camada DWS dws_users e dws_shops.

   No console do Realtime Compute for Apache Flink, acesse a página Data Development > Data Exploration. Na aba Query Editor, copie o código a seguir para o editor, selecione a instrução e clique em Run no canto superior direito.

   SELECT
     order_create_time AS order_create_date,
     order_product_catalog_name,
     COUNT(*),
     SUM(order_fee)
   FROM
     paimoncatalog.order_dw.dwd_orders
   WHERE
     order_create_time >= '2023-02-01 00:00:00'  and order_create_time < '2023-03-01 00:00:00'
   GROUP BY
     order_create_date, order_product_catalog_name
   ORDER BY
     order_create_date, order_product_catalog_name;

   A mensagem The following statement has been executed successfully! indica que a criação foi bem-sucedida.

2. Consuma dados de alteração da tabela dwd_orders da camada DWD.

   No console do Realtime Compute for Apache Flink, acesse a página Data Development > ETL, crie um novo job de streaming SQL chamado dwm. Copie o código a seguir para o editor SQL, clique em Deploy e inicie o job usando Stateless Start.

   Este job SQL grava dados da tabela dwd_orders na tabela dwm_users_shops. O mecanismo de mesclagem de dados de pré-agregação do Paimon soma automaticamente o order_fee para calcular o valor total gasto por um usuário em uma loja. Ele também soma o valor 1 para calcular o número de vezes que um usuário fez uma compra nessa loja.

   SELECT
     order_create_time AS order_create_date,
     order_product_catalog_name,
     COUNT(*),
     SUM(order_fee)
   FROM
     paimoncatalog.order_dw.dwd_orders
   WHERE
     order_create_time >= '2023-02-01 00:00:00'  and order_create_time < '2023-03-01 00:00:00'
   GROUP BY
     order_create_date, order_product_catalog_name
   ORDER BY
     order_create_date, order_product_catalog_name;

3. Consuma dados de alteração da tabela dwm_users_shops da camada DWM em tempo real.

   No console do Realtime Compute for Apache Flink, acesse a página Data Development > ETL. Crie um novo job de streaming SQL chamado dws. Copie o código a seguir para o editor SQL, clique em Deploy e inicie o job usando Stateless Start.

   Este job SQL grava dados da tabela dwm_users_shops nas tabelas dws_users e dws_shops. O mecanismo de mesclagem de dados de pré-agregação do Paimon é usado para calcular o gasto total de cada usuário (paid_buy_fee_sum) na tabela dws_users. Na tabela dws_shops, ele calcula as vendas totais da loja (paid_buy_fee_sum), o número de usuários compradores (somando 1) e o número total de compras (pv).

   SELECT
     order_create_time AS order_create_date,
     order_product_catalog_name,
     COUNT(*),
     SUM(order_fee)
   FROM
     paimoncatalog.order_dw.dwd_orders
   WHERE
     order_create_time >= '2023-02-01 00:00:00'  and order_create_time < '2023-03-01 00:00:00'
   GROUP BY
     order_create_date, order_product_catalog_name
   ORDER BY
     order_create_date, order_product_catalog_name;

4. Visualize os dados nas tabelas dws_users e dws_shops.

   No console do Realtime Compute for Apache Flink, acesse a página Data Development > Data Exploration. Na aba Query Editor, copie o código a seguir para o editor, selecione a instrução e clique em Run no canto superior direito.

   SELECT
     order_create_time AS order_create_date,
     order_product_catalog_name,
     COUNT(*),
     SUM(order_fee)
   FROM
     paimoncatalog.order_dw.dwd_orders
   WHERE
     order_create_time >= '2023-02-01 00:00:00'  and order_create_time < '2023-03-01 00:00:00'
   GROUP BY
     order_create_date, order_product_catalog_name
   ORDER BY
     order_create_date, order_product_catalog_name;

   Após a execução da consulta, a tabela dws_users retorna três registros com as seguintes colunas: user_id, ds e paid_buy_fee_sum. Os registros são: user_001 / 20230215 / 8000, user_002 / 20230215 / 5000 e user_003 / 20230215 / 5000.

   SELECT
     order_create_time AS order_create_date,
     order_product_catalog_name,
     COUNT(*),
     SUM(order_fee)
   FROM
     paimoncatalog.order_dw.dwd_orders
   WHERE
     order_create_time >= '2023-02-01 00:00:00'  and order_create_time < '2023-03-01 00:00:00'
   GROUP BY
     order_create_date, order_product_catalog_name
   ORDER BY
     order_create_date, order_product_catalog_name;

   Ao consultar a tabela dws_shops, o resultado contém quatro linhas e cinco colunas: shop_id, ds, paid_buy_fee_sum, uv e pv. Os shop_ids são 12345 a 12348, ds é 20230215 para todos os registros, os valores de paid_buy_fee_sum são 5000, 4000, 7000 e 2000, os valores de uv são 1, 1, 2 e 2, e os valores de pv são 1, 1, 3 e 2, respectivamente.

Capturar alterações do banco de dados de negócios

Após construir o streaming data lakehouse, teste sua capacidade de capturar alterações do banco de dados de negócios.

1. Insira os seguintes dados no banco de dados order_dw no MySQL.

   SELECT
     order_create_time AS order_create_date,
     order_product_catalog_name,
     COUNT(*),
     SUM(order_fee)
   FROM
     paimoncatalog.order_dw.dwd_orders
   WHERE
     order_create_time >= '2023-02-01 00:00:00'  and order_create_time < '2023-03-01 00:00:00'
   GROUP BY
     order_create_date, order_product_catalog_name
   ORDER BY
     order_create_date, order_product_catalog_name;

2. Visualize os dados nas tabelas dws_users e dws_shops. No console do Realtime Compute for Apache Flink, acesse a página Data Development > Data Exploration. Na aba Query Editor, copie o código a seguir para o editor, selecione a instrução e clique em Run no canto superior direito.

   • Tabela dws_users

     SELECT
       order_create_time AS order_create_date,
       order_product_catalog_name,
       COUNT(*),
       SUM(order_fee)
     FROM
       paimoncatalog.order_dw.dwd_orders
     WHERE
       order_create_time >= '2023-02-01 00:00:00'  and order_create_time < '2023-03-01 00:00:00'
     GROUP BY
       order_create_date, order_product_catalog_name
     ORDER BY
       order_create_date, order_product_catalog_name;

     A consulta retorna três linhas com as seguintes colunas: user_id, ds e paid_buy_fee_sum. Os registros são: user_001 / 20230215 / 11000, user_002 / 20230215 / 6000 e user_003 / 20230215 / 7000.

   • Tabela dws_shops

     SELECT
       order_create_time AS order_create_date,
       order_product_catalog_name,
       COUNT(*),
       SUM(order_fee)
     FROM
       paimoncatalog.order_dw.dwd_orders
     WHERE
       order_create_time >= '2023-02-01 00:00:00'  and order_create_time < '2023-03-01 00:00:00'
     GROUP BY
       order_create_date, order_product_catalog_name
     ORDER BY
       order_create_date, order_product_catalog_name;

     A consulta retorna quatro linhas com cinco colunas: shop_id, ds, paid_buy_fee_sum, uv e pv. Os shop_ids são 12345, 12346, 12347 e 12348, ds é 20230215 para todos os registros, os valores de paid_buy_fee_sum são 8000, 4000, 7000 e 5000, os valores de uv são 1, 1, 2 e 3, e os valores de pv são 2, 1, 3 e 4, respectivamente.