Realtime Compute for Apache Flink:Crie um data warehouse em tempo real com o Hologres

Crie uma instância do RDS for MySQL e prepare os dados

1. Crie uma instância do ApsaraDB RDS for MySQL. Para mais informações, consulte Create an ApsaraDB RDS for MySQL instance.

   A instância do ApsaraDB RDS for MySQL deve estar na mesma VPC do seu workspace do Flink e da instância do Hologres.

2. Crie um banco de dados e uma conta.

   Para a instância de destino, crie um banco de dados chamado order_dw e uma conta padrão com permissões de leitura e gravação no banco de dados. Para mais informações, consulte Create a database e Create an account.

3. Prepare a source de dados MySQL CDC.

   1. Na página de detalhes da instância, clique em Log On to Database.

   2. Na página de logon, insira o nome de usuário e a senha da conta do banco de dados criada e, em seguida, clique em Log On.

   3. Após fazer o logon, clique duas vezes no banco de dados order_dw para acessá-lo.

   4. No SQL Console, insira as seguintes instruções DDL para criar as tabelas de negócios e as instruções INSERT para preenchê-las.

      CREATE TABLE `orders` (
        order_id bigint not null primary key,
        user_id varchar(50) not null,
        shop_id bigint not null,
        product_id bigint not null,
        buy_fee numeric(20,2) not null,   
        create_time timestamp not null,
        update_time timestamp not null default now(),
        state int not null 
      );
      CREATE TABLE `orders_pay` (
        pay_id bigint not null primary key,
        order_id bigint not null,
        pay_platform int not null,
        create_time timestamp not null
      );
      CREATE TABLE `product_catalog` (
        product_id bigint not null primary key,
        catalog_name varchar(50) not null
      );
      -- Prepare data.
      INSERT INTO product_catalog VALUES(1, 'phone_aaa'),(2, 'phone_bbb'),(3, 'phone_ccc'),(4, 'phone_ddd'),(5, 'phone_eee');
      INSERT INTO orders VALUES
      (100001, 'user_001', 12345, 1, 5000.05, '2023-02-15 16:40:56', '2023-02-15 18:42:56', 1),
      (100002, 'user_002', 12346, 2, 4000.04, '2023-02-15 15:40:56', '2023-02-15 18:42:56', 1),
      (100003, 'user_003', 12347, 3, 3000.03, '2023-02-15 14:40:56', '2023-02-15 18:42:56', 1),
      (100004, 'user_001', 12347, 4, 2000.02, '2023-02-15 13:40:56', '2023-02-15 18:42:56', 1),
      (100005, 'user_002', 12348, 5, 1000.01, '2023-02-15 12:40:56', '2023-02-15 18:42:56', 1),
      (100006, 'user_001', 12348, 1, 1000.01, '2023-02-15 11:40:56', '2023-02-15 18:42:56', 1),
      (100007, 'user_003', 12347, 4, 2000.02, '2023-02-15 10:40:56', '2023-02-15 18:42:56', 1);
      INSERT INTO orders_pay VALUES
      (2001, 100001, 1, '2023-02-15 17:40:56'),
      (2002, 100002, 1, '2023-02-15 17:40:56'),
      (2003, 100003, 0, '2023-02-15 17:40:56'),
      (2004, 100004, 0, '2023-02-15 17:40:56'),
      (2005, 100005, 0, '2023-02-15 18:40:56'),
      (2006, 100006, 0, '2023-02-15 18:40:56'),
      (2007, 100007, 0, '2023-02-15 18:40:56');

4. Clique em Execute e, em seguida, clique em Direct Execution.

Crie uma instância do Hologres e compute groups

1. Adquira uma instância dedicada do Hologres. Para mais informações, consulte Purchase a Hologres instance.

   A instância do Hologres deve estar na mesma VPC da instância do ApsaraDB RDS for MySQL. Para experimentar o forte isolamento de recursos por meio da divisão de leitura/gravação, este exemplo usa o tipo de instância Virtual Warehouse e define Reserved Compute Resource como 64 para permitir a criação de compute groups adicionais.

2. Após log on to the instance, crie um banco de dados e conceda permissões.

   Crie um banco de dados chamado order_dw (com o modelo de permissão simples ativado) e conceda privilégios administrativos ao usuário. Para detalhes sobre como gerencie e autorize o banco de dados, consulte Manage databases.

   Nota

   • Caso não encontre a conta na lista suspensa User, significa que a conta não foi adicionada à instância. acesse a página User Management e adicione o usuário como SuperUser.

   • No Hologres V2.0 e posterior, a extensão de binary logging é ativada por padrão. Não é necessário ativá-la manualmente.

3. Crie um novo compute group.

   Use diferentes compute groups para isolar recursos. Use o compute group inicial init_warehouse para gravações de dados e o compute group read_warehouse_1 para atender a consultas.

   Todos os recursos de computação reservados são alocados para o compute group inicial init_warehouse por padrão. É necessário reduzir primeiro seus recursos antes de criar um novo compute group. Para mais informações, consulte Create a new compute group instance.

   1. Acesse Security > Compute Group Management e confirme o nome da instância.

   2. Na linha do compute group init_warehouse, clique em Modify Configuration na coluna Actions. Reduza os recursos alocados e clique em OK.

   3. Clique em Create Compute Group, crie um novo compute group chamado read_warehouse_1 e, em seguida, clique em OK.

Crie um workspace e catálogos do Flink

1. Crie um workspace do Flink. Para mais informações, consulte Activate Realtime Compute for Apache Flink.

   O workspace do Flink deve estar na mesma VPC das instâncias do ApsaraDB RDS for MySQL e do Hologres.

2. Acesse o Realtime Compute for Apache Flink console e clique em Console na coluna Actions do seu workspace.

3. Crie um cluster de sessão para fornecer um ambiente de execução para criar catálogos e executar scripts. Para mais informações, consulte Step 1: Create a session cluster.

4. Crie um catálogo do Hologres.

   Na página Development > Scripts, na aba Scripts, copie o código a seguir, substitua os valores dos placeholders, selecione o código e, em seguida, clique em Run. Essa ação utiliza o cluster de sessão criado como ambiente de execução.

   CREATE CATALOG dw WITH (
     'type' = 'hologres',
     'endpoint' = '<ENDPOINT>', 
     'username' = 'BASIC$flinktest',
     'password' = '${secret_values.holosecrect}',
     'dbname' = 'order_dw@init_warehouse', -- Specify the database name and connect to the init_warehouse compute group.
     'binlog' = 'true', -- You can set default WITH options for source, dimension, and result tables when creating the catalog. Tables created under this catalog inherit these defaults.
     'sdkMode' = 'jdbc', -- The jdbc mode is recommended.
     'cdcmode' = 'true',
     'connectionpoolname' = 'the_conn_pool',
     'ignoredelete' = 'true',  -- Required for wide-table merge to prevent retractions.
     'partial-insert.enabled' = 'true', -- Required for wide-table merge to enable partial column updates.
     'mutateType' = 'insertOrUpdate', -- Required for wide-table merge to enable partial column updates.
     'table_property.binlog.level' = 'replica', -- You can also pass persistent Hologres table properties when creating the catalog. Tables created later will have binary logging enabled by default.
     'table_property.binlog.ttl' = '259200'
   );

   Modifique os seguintes parâmetros com as informações reais do seu service do Hologres.

   Parâmetro	Descrição	Notas
   endpoint	O endpoint da sua instância do Hologres.	Na página de detalhes da instância do Hologres, obtenha o nome de domínio para a VPC que você especifique. Para mais informações sobre nomes de domínio, consulte Endpoints.
   username	Selecione uma das seguintes opções: O nome de usuário para uma conta personalizada deve ser formatado como BASIC$<user_name>. O AccessKey ID da sua conta Alibaba Cloud ou usuário RAM.	O usuário configurado deve ter acesso ao banco de dados Hologres correspondente. Para detalhes, consulte Hologres permission model e Manage users. Este exemplo usa uma conta personalizada chamada BASIC$flinktest e define sua senha por meio de uma variável de projeto chamada holosecrect para evitar os riscos de segurança do armazenamento de senhas em texto simples. Para mais informações, consulte Project variables.
   password	A senha da conta personalizada. O AccessKey secret da sua conta Alibaba Cloud ou usuário RAM.

   Nota

   Ao criar um catálogo, é possível definir opções WITH padrão para tabelas source, dimensão e resultado. Também é possível definir propriedades padrão para tabelas físicas do Hologres, como os parâmetros que começam com table_property. Para mais informações, consulte Manage Hologres catalogs e Hologres connector for real-time data warehouses (WITH parameters).

Crie a camada ODS: ingira dados de negócios

Com um Flink CDC data ingestion job, é possível usar o recurso full database synchronization para criar a camada ODS em uma única etapa. A camada ODS normalmente atua como uma fonte de eventos para jobs de streaming, em vez de ser consultada diretamente para consultas OLAP ou point queries. Ative o binary logging, pois é suficiente para esse propósito. O binary logging é um recurso principal do Hologres. O conector do Hologres também suporta um modo full-plus-incremental: ele lê um snapshot completo primeiro e, em seguida, consome binary logs incrementalmente.

1. Crie o job YAML de ingestão de dados ODS.

   1. Na página Development > Data Ingestion, crie um rascunho YAML de ingestão de dados chamado ODS e, em seguida, copie o código a seguir no editor YAML. Recomenda-se selecionar VVR 11.1 como a versão do motor. Para utilizar VVR 8.x, selecione VVR 8.0.11. Para mais informações, consulte Develop a Flink CDC data ingestion job.

      source:
        type: mysql
        name: MySQL Source
        # The internal endpoint of the MySQL instance.
        hostname: <hostname>
        port: 3306
        username: <username>
        password: ${secret_values.mysql_pw}
        # Synchronize all tables in the order_dw database. You can specify the tables to ingest as needed.
        tables: order_dw.\.*
        # The server-id range of the MySQL CDC instance.
        server-id: 8001-8004
      
      sink:
        type: hologres
        name: Hologres Sink
        endpoint: <ENDPOINT>
        dbname: order_dw
        username: BASIC$flinktest
        password: ${secret_values.holosecrect}
        # Enable binary logging when the tables are created, so that the downstream DWD job can consume the binary logs.
        table_property.binlog.level: replica
        table_property.binlog.ttl: 259200
      
      pipeline:
        name: MySQL to Hologres ODS

      Modifique os seguintes parâmetros da source com as informações reais do seu service MySQL.

      Parâmetro	Descrição
      hostname	O endereço IP ou hostname do seu banco de dados MySQL. Na página de informações básicas do banco de dados, clique em View Connection Details na área Network Type para obter o endpoint interno.
      port	O número da porta do seu service de banco de dados MySQL. O valor padrão é 3306.
      username	O nome de usuário do seu service de banco de dados MySQL.
      password	A senha do seu service de banco de dados MySQL. Este exemplo usa uma variável chamada mysql_pw para a senha, para evitar a exposição em texto simples. Para mais informações, consulte Manage variables.

      O endpoint, o nome de usuário e a senha do sink são os mesmos do catálogo do Hologres criado na Etapa 1. Defina dbname como o nome do banco de dados Hologres de destino. Caso o dbname não especifique um compute group, o Hologres roteia as requisições de gravação para o compute group padrão da instância, que é o order_dw neste exemplo. Este também é o compute group utilizado pelo catálogo do Hologres para gravações.

      Nota

      • Por padrão, este exemplo sincroniza dados para o schema Public do banco de dados read_warehouse_1. O job cria automaticamente as tabelas de resultado, portanto, não é necessário criá-las com antecedência. Para sincronizar dados para um schema que você especifique no banco de dados Hologres de destino, configure o mapeamento entre as tabelas source e as tabelas de sink na seção route. Para mais informações, consulte Synchronizing to a specified schema.

      • Alterações no schema da tabela nas tabelas source, como adição, exclusão e renomeação de colunas, são sincronizadas com as tabelas de resultado com base na política LENIENT padrão. Para mais informações sobre como cada tipo de alteração é tratada, consulte Table schema synchronization.

   2. No canto superior direito, clique em Deploy para implantar o job.

   3. No painel de navegação à esquerda, escolha O&M > Deployments. Na linha do job ODS recém-implantado, clique em Start na coluna Actions. Selecione Initial Mode e, em seguida, clique em Start.

2. Carregue os dados no compute group.

   Um table group é uma unidade de armazenamento no Hologres. Ao usar o compute group read_warehouse_1 para consultar dados de um table group no banco de dados order_dw, como order_dw_tg_default (para criar um table group, consulte Table Group Management), o table group order_dw_tg_default é carregado para o compute group read_warehouse_1. Isso permite usar o compute group init_warehouse para gravar dados e o compute group read_warehouse_1 para consultas de service.

   Na página de desenvolvimento do HoloWeb, clique em SQL Editor. Confirme o nome da instância e o nome do banco de dados e, em seguida, execute os seguintes comandos. Para mais informações, consulte Create a new compute group instance. Após o carregamento, verifique que o order_dw_tg_default carregou os dados do table group read_warehouse_1.

   -- List table groups in the current database.
   SELECT tablegroup_name FROM hologres.hg_table_group_properties GROUP BY tablegroup_name;
   -- Load a table group into a compute group.
   CALL hg_table_group_load_to_warehouse ('order_dw.order_dw_tg_default', 'read_warehouse_1', 1);
   -- Check the table groups loaded into the compute group.
   select * from hologres.hg_warehouse_table_groups;

3. No canto superior direito, mude o compute group para read_warehouse_1. As consultas e análises subsequentes utilizarão este compute group.

   No canto superior direito da página do HoloWeb, selecione product_catalog na lista suspensa do compute group.

4. Na página SQL Editor, execute os seguintes comandos para visualizar os dados sincronizados do MySQL para as três tabelas do Hologres.

   -- Query data from the orders table.
   SELECT * FROM orders;
   -- Query data from the orders_pay table.
   SELECT * FROM orders_pay;
   -- Query data from the product_catalog table.
   SELECT * FROM product_catalog;

   O resultado da consulta para a tabela dwd_orders contém duas colunas, product_id (1 a 5) e catalog_name (phone_aaa, phone_bbb, phone_ccc, phone_ddd, phone_eee), com um total de 5 registros. Isso confirma a sincronização bem-sucedida dos dados para o Hologres.

Crie a camada DWD: crie uma wide table em tempo real

Esta etapa usa o recurso de atualização parcial de colunas do conector do Hologres. É possível expressar atualizações parciais com INSERT DML. O job consulta várias tabelas de dimensões por meio de point queries de alto desempenho, viabilizadas pelo armazenamento em linhas e pelo armazenamento híbrido de linhas e colunas do Hologres. Com o forte isolamento de recursos, as cargas de trabalho de gravação, leitura e análise não interferem umas nas outras.

1. Use o recurso de catálogo do Flink para criar a wide table da camada DWD orders no Hologres.

   Na página Development > Scripts, copie o código a seguir na aba Scripts, selecione o código e, em seguida, clique em Run.

   -- Wide table columns must be nullable because different streams write to the same result table, and any column can be null.
   CREATE TABLE dw.order_dw.dwd_orders (
     order_id bigint not null,
     order_user_id string,
     order_shop_id bigint,
     order_product_id bigint,
     order_product_catalog_name string,
     order_fee numeric(20,2),
     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,
     PRIMARY KEY(order_id) NOT ENFORCED
   );
   -- You can modify Hologres physical table properties through the catalog.
   ALTER TABLE dw.order_dw.dwd_orders SET (
     'table_property.binlog.ttl' = '604800' -- Change the binary log TTL to one week.
   );

2. Consuma alterações de binary logging das tabelas da camada ODS orders_pay e orders em tempo real.

   Na página Development > ETL, crie um novo rascunho de stream SQL chamado DWD. Copie o código a seguir no editor SQL e, em seguida, clique em Deploy e Start para o job. Este job SQL faz o join das tabelas product_catalog e dwd_orders por meio de um temporal join e grava o resultado na tabela dwd_orders, enriquecendo os dados em tempo real.

   BEGIN STATEMENT SET;
   INSERT INTO dw.order_dw.dwd_orders 
    (
      order_id,
      order_user_id,
      order_shop_id,
      order_product_id,
      order_fee,
      order_create_time,
      order_update_time,
      order_state,
      order_product_catalog_name
    ) SELECT o.*, dim.catalog_name 
      FROM dw.order_dw.orders as o
      LEFT JOIN dw.order_dw.product_catalog FOR SYSTEM_TIME AS OF proctime() AS dim
      ON o.product_id = dim.product_id;
   INSERT INTO dw.order_dw.dwd_orders 
     (pay_id, order_id, pay_platform, pay_create_time)
      SELECT * FROM dw.order_dw.orders_pay;
   END;

3. e visualize os dados na wide table dws_users.

   Conecte-se à instância do Hologres na página de desenvolvimento do HoloWeb, faça logon no banco de dados de destino e, em seguida, execute o seguinte comando no SQL Editor.

   SELECT * FROM dwd_orders;

   Após a execução bem-sucedida, o resultado da consulta retorna dados da wide table dwd_orders, incluindo campos como order_id, order_user_id, order_shop_id, order_product_id, order_product_catalog_name, order_fee, order_create_time e order_update_time.

Crie a camada DWS: calcule métricas em tempo real

1. Use o recurso de catálogo do Flink para criar as tabelas agregadas da camada DWS dws_shops e dw.order_dw.dwd_orders no Hologres.

   Na página Development > Scripts, copie o código a seguir na aba Scripts, selecione o código e, em seguida, clique em Run.

   -- User-dimension aggregate table.
   CREATE TABLE dw.order_dw.dws_users (
     user_id string not null,
     ds string not null,
     paied_buy_fee_sum numeric(20,2) not null comment 'Total amount of payments completed on the day',
     primary key(user_id, ds) NOT ENFORCED
   );
   -- Shop-dimension aggregate table.
   CREATE TABLE dw.order_dw.dws_shops (
     shop_id bigint not null,
     ds string not null,
     paied_buy_fee_sum numeric(20,2) not null comment 'Total amount of payments completed on the day',
     primary key(shop_id, ds) NOT ENFORCED
   );

2. Consuma a wide table da camada DWD user_001 em tempo real, agregue dados no Flink e grave os resultados finais nas tabelas DWS no Hologres.

   Na página Development > ETL, crie um novo rascunho de stream SQL chamado DWS. Copie o código a seguir no editor SQL e, em seguida, clique em Deploy e Start para o job.

   BEGIN STATEMENT SET;
   INSERT INTO dw.order_dw.dws_users
     SELECT 
       order_user_id,
       DATE_FORMAT (pay_create_time, 'yyyyMMdd') as ds,
       SUM (order_fee)
       FROM dw.order_dw.dwd_orders c
       WHERE pay_id IS NOT NULL AND order_fee IS NOT NULL -- Both order and payment stream data have been written to the wide table.
       GROUP BY order_user_id, DATE_FORMAT (pay_create_time, 'yyyyMMdd');
   INSERT INTO dw.order_dw.dws_shops
     SELECT 
       order_shop_id,
       DATE_FORMAT (pay_create_time, 'yyyyMMdd') as ds,
       SUM (order_fee)
      FROM dw.order_dw.dwd_orders c
      WHERE pay_id IS NOT NULL AND order_fee IS NOT NULL -- Both order and payment stream data have been written to the wide table.
      GROUP BY order_shop_id, DATE_FORMAT (pay_create_time, 'yyyyMMdd');
   END;

3. Visualize os resultados agregados na camada DWS. Os resultados são atualizados em tempo real conforme os dados upstream mudam.

   1. Visualize os dados no console do Hologres antes da alteração.

      dws_users table

      SELECT * FROM dws_users;

      Após executar a consulta, o resultado retorna dados da tabela dws_users, que inclui três colunas: user_id, ds e paied_buy_fee_sum. No resultado do exemplo, a coluna user_id contém os valores user_002, user_003 e 20230215; a coluna ds contém o valor 8000.08; e a coluna paied_buy_fee_sum contém os valores 5000.05, 5000.05 e dws_shops, respectivamente. A coluna user_id identifica exclusivamente cada usuário.

      dws_shops table

      SELECT * FROM dws_shops;

      O resultado da consulta mostra que a tabela orders contém três colunas: shop_id (ID da loja), ds (partição de data) e paied_buy_fee_sum (valor do pagamento). O resultado retorna quatro linhas de dados de exemplo, confirmando a criação bem-sucedida da tabela da camada DWS.

   2. No console do RDS, insira um novo registro em cada uma das tabelas orders_pay e order_dw no banco de dados dwd_orders.

      INSERT INTO orders VALUES
      (100008, 'user_003', 12345, 5, 6000.02, '2023-02-15 09:40:56', '2023-02-15 18:42:56', 1);
      INSERT INTO orders_pay VALUES
      (2008, 100008, 1, '2023-02-15 19:40:56');

   3. Visualize os dados atualizados no console do Hologres.

      dwd_orders table

      SELECT * FROM dwd_orders;

      Após executar a consulta, a tabela order_id=100008 retorna oito registros de pedidos, incluindo campos como 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 oitavo registro (user_003, phone_eee, 6000.02, user_001 / 20230215 / 8000.08) corresponde aos dados recém-gravados.

      dws_users table

      SELECT * FROM dws_users;

      A consulta retorna três linhas com três colunas: user_id, ds e paied_buy_fee_sum: user_002 / 20230215 / 5000.05, user_003 / 20230215 / 11000.07 e user_001. O maior valor total de pagamento é o do user_003 (11000.07).

      dws_shops table

      SELECT * FROM dws_shops;

      Após executar a consulta, o resultado contém três colunas, shop_id, ds e paied_buy_fee_sum, com quatro linhas de dados mostrando os valores de taxa (11000.07, 4000.04, 7000.07 e 2000.02) para as lojas 12345, 12346, 12347 e 12348 em 20230215. shop_id e paied_buy_fee_sum são as colunas de métricas principais.

Analise os dados

Como o binary logging está ativado, é possível inspecionar diretamente as alterações de dados. Caso seja necessário explorar dados de negócios em resultados intermediários ou verificar a correção da computação final, cada camada desta solução é persistida, facilitando o exame do processo intermediário.

SELECT *
FROM dw.order_dw.dwd_orders /*+ OPTIONS('binlog'='false') */ 
WHERE order_user_id = 'user_001' and order_create_time > '2023-02-15 12:00:00'; -- Batch mode supports filter pushdown to improve batch job execution efficiency.

Point query

Consulte as tabelas de métricas agregadas da camada DWS por chave primária, com suporte para milhões de RPS.

Na página de desenvolvimento do HoloWeb, execute o seguinte SQL para consultar o valor de consumo de um usuário específico em uma data específica.

-- holo sql
SELECT * FROM dws_users WHERE user_id ='user_001' AND ds = '20230215';

O resultado da consulta retorna três colunas: ds, paied_buy_fee_sum e user_001 (valor de consumo). O valor de consumo para 20230215 em 8000.08 é order_id.

Consulta OLAP

Execute consultas OLAP na wide table da camada DWD.

Na página de desenvolvimento do HoloWeb, execute o seguinte SQL para consultar os detalhes do pedido de um cliente específico em uma plataforma de pagamento específica em fevereiro de 2023.

-- holo sql
SELECT * FROM dwd_orders
WHERE order_create_time >= '2023-02-01 00:00:00'  and order_create_time < '2023-03-01 00:00:00'
AND order_user_id = 'user_001'
AND pay_platform = 0
ORDER BY order_create_time LIMIT 100;

O resultado da consulta retorna dois registros de pedidos com campos como order_user_id, order_shop_id, order_product_id, order_fee, order_create_time, order_update_time e 20230215. Os IDs de pedido do exemplo são 100006 e 100004.

Relatórios em tempo real

Gere relatórios em tempo real com base em dados da wide table da camada DWD. O armazenamento híbrido de linhas e colunas e as tabelas orientadas a colunas no Hologres fornecem excelentes recursos de consulta OLAP, garantindo respostas em segundos.

Na página de desenvolvimento do HoloWeb, execute o seguinte SQL para consultar o número total de pedidos e o valor total dos pedidos para cada categoria de product em fevereiro de 2023.

-- holo sql
SELECT
  TO_CHAR(order_create_time, 'YYYYMMDD') AS order_create_date,
  order_product_catalog_name,
  COUNT(*),
  SUM(order_fee)
FROM
  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 executar a instrução SQL, a aba Result exibe uma tabela com quatro colunas: order_create_date, order_product_catalog_name, count e sum. Por exemplo, os resultados para a data phone_aaa mostram que a categoria phone_bbb teve 2 pedidos totalizando 6000.06, a categoria phone_bbb teve 1 pedido de 4000.04, e assim por diante para todas as cinco categorias.