Realtime Compute for Apache Flink:Materialized Table (Build a Unified Stream-Batch Lakehouse)

1. Crie um Paimon Catalog.

   As materialized tables são viabilizadas pelo Apache Paimon. É necessário criar um Paimon Catalog com o tipo de metastore Filesystem . Pule esta etapa se você já tiver um. Para mais informações, consulte Create a Paimon Catalog.

   Create a Paimon Catalog

   1. Acesse o Realtime Compute Management Console.

   2. Clique em Console na coluna Actions do workspace desejado.

   3. No painel de navegação à esquerda, escolha Catalogs e clique em Create Catalog. Selecione Apache Paimon e clique em Next.

      Descrição dos parâmetros:

      Item de configuração	Descrição	Observações
      metastore	Tipo de metastore.	Este exemplo usa filesystem como tipo de metastore.
      catalog name	Nome do Paimon Catalog.	Insira um nome em inglês personalizado. Este exemplo usa paimon.
      warehouse	Diretório do data warehouse no OSS.	Use o formato oss://<bucket>/<object>. Onde: <bucket>: o nome do seu bucket OSS. <object>: o caminho onde seus dados estão armazenados. Verifique os nomes do bucket e do object no OSS Management Console.
      fs.oss.endpoint	Endereço de conexão do OSS.	Se o Flink e o OSS estiverem na mesma região, use o endpoint de rede privada. Caso contrário, use o endpoint de rede pública. Para mais informações, consulte Regions and endpoints.
      fs.oss.accessKeyId	AccessKey ID da sua conta Alibaba Cloud ou usuário RAM com permissões de leitura e gravação no OSS.	Para obter instruções, consulte Create an AccessKey. Para evitar expor credenciais em texto simples, use variáveis. Para mais informações, consulte Variable management.
      fs.oss.accessKeySecret	AccessKey secret da sua conta Alibaba Cloud ou usuário RAM com permissões de leitura e gravação no OSS.

2. Crie a tabela de log comportamental do usuário ods_user_log e a tabela de informações de produtos ods_dim_product.

   1. Acesse o Realtime Compute Management Console.

   2. Clique em Console na coluna Actions do workspace desejado.

   3. No painel de navegação à esquerda, escolha Development > Scripts. Copie e cole o seguinte código para criar as tabelas source.

      Este exemplo assume que você já criou um Paimon Catalog chamado paimon e está usando o banco de dados padrão (default).

      CREATE TABLE `paimon`.`default`.`ods_user_log` (
        item_id INT NOT NULL,
        user_id INT NOT NULL,
        vtime TIMESTAMP(6),
        ds VARCHAR(10) NOT NULL
      ) 
      PARTITIONED BY(ds)
      WITH (
        'bucket' = '4',            -- Set the number of buckets to 4.
        'bucket-key' = 'item_id'   -- Specify the key used to determine bucket assignment. Rows with the same item_id go into the same bucket.
      );
      CREATE TABLE `paimon`.`default`.`ods_dim_product` (
        item_id INT NOT NULL,
        title VARCHAR(255),
        pict_url VARCHAR(255), 
        brand_id INT,
        seller_id INT,
        PRIMARY KEY(item_id) NOT ENFORCED
      ) WITH (
        'bucket' = '4',
        'bucket-key' = 'item_id'
      );

   4. Clique em Run no canto superior direito para criar as tabelas.

   5. No painel de navegação à esquerda, escolha Catalogs, clique no seu Paimon Catalog e, em seguida, clique em Refresh para visualizar as novas tabelas.

3. Use o Faker connector for simulated data generation para gerar dados simulados e gravá-los nas tabelas Paimon.

   1. No painel de navegação à esquerda, escolha Development > ETL.

   2. Clique em New, selecione Blank stream draft, clique em Next e, em seguida, clique em Create.

   3. Copie a seguinte instrução SQL para o editor SQL.

      CREATE TEMPORARY TABLE `user_log` (
        item_id INT,  // Product ID
        user_id INT,  // User ID
        vtime TIMESTAMP,  
        ds AS DATE_FORMAT(CURRENT_DATE,'yyyyMMdd')
      ) WITH (
        'connector' = 'faker',    -- Faker connector
        'fields.item_id.expression'='#{number.numberBetween ''0'',''1000''}',    -- Generate a random number between 0 and 1000
        'fields.user_id.expression'='#{number.numberBetween ''0'',''100''}',
        'fields.vtime.expression'='#{date.past ''5'',''HOURS''}',           -- Generate data up to 5 hours before the current time
        'rows-per-second' = '3'   -- Generate 3 rows per second
       );
       CREATE TEMPORARY TABLE `dim_product` (
        item_id INT NOT NULL,
        title VARCHAR(255),
        pict_url VARCHAR(255), 
        brand_id INT,
        seller_id INT,
        PRIMARY KEY(item_id) NOT ENFORCED
       ) WITH (
        'connector' = 'faker',    -- Faker connector
        'fields.item_id.expression'='#{number.numberBetween ''0'',''1000''}',
        'fields.title.expression'='#{book.title}',
        'fields.pict_url.expression'='#{internet.domainName}',
        'fields.brand_id.expression'='#{number.numberBetween ''1000'',''10000''}',   
        'fields.seller_id.expression'='#{number.numberBetween ''1000'',''10000''}',
        'rows-per-second' = '3'        -- Generate 3 rows per second
       );
      BEGIN STATEMENT SET; 
      INSERT INTO `paimon`.`default`.`ods_user_log` 
        SELECT 
        item_id,
        user_id,
        vtime,
        CAST(ds AS VARCHAR(10)) AS ds
      FROM `user_log`;
      INSERT INTO `paimon`.`default`.`ods_dim_product`
        SELECT 
        item_id,
        title,
        pict_url,
        brand_id,
        seller_id
      FROM `dim_product`;
      END; 

   4. Clique em Deploy no canto superior direito para implantar o job.

   5. No painel de navegação à esquerda, escolha O&M > Deployments. Clique em Start na coluna Actions do job desejado, selecione Initial Mode e clique em Start.

4. Consulte os dados simulados.

   No painel de navegação à esquerda, escolha Development > Scripts. Copie a seguinte instrução SQL para o editor SQL e clique em Run no canto superior direito.

   SELECT * FROM `paimon`.`default`.ods_dim_product LIMIT 10;
   SELECT * FROM `paimon`.`default`.ods_user_log LIMIT 10;

Prévia de dados

É possível visualizar as últimas 100 linhas de uma materialized table.

1. No painel de navegação à esquerda, escolha O&M > Data Lineage e pesquise pela materialized table desejada.

2. Clique na view da materialized table e, em seguida, clique em Details no canto inferior direito da página.

3. Na aba Data Preview da materialized table, clique no ícone Query.

Consulta de dados

No painel de navegação à esquerda, escolha Development > Scripts. Copie a seguinte instrução SQL para o editor SQL, selecione o trecho de código e clique em Run para consultar a materialized table dws_overall.

SELECT * FROM `paimon`.`default`.dws_overall ORDER BY hh;