Realtime Compute for Apache Flink:Análise de comportamento do usuário com Flink, MongoDB e Hologres

Fluxo de propagação de alterações

Cada alteração passa por quatro etapas:

1. Captura: Detecta alterações em tempo real nas tabelas de dimensões do MongoDB.

2. Propagação: Ao detectar mudança em uma tabela de dimensões, o Job 1 realiza um lookup join (por exemplo, em game_id) para localizar as linhas afetadas na tabela de fatos e extrai suas PKs (por exemplo, sale_id).

3. Disparo: Envia as PKs para o Kafka para notificar o Job 2 sobre atualizações pendentes.

4. Upsert: O Job 2 busca os dados mais recentes, reconstrói a linha da tabela larga e executa o upsert no Hologres.

Criar coleções do MongoDB

1. Log on to your ApsaraDB for MongoDB instance.

2. Adicione o bloco CIDR do seu workspace Flink à lista de permissões do MongoDB. Para detalhes, consulte Configure a whitelist for an instance e How do I configure a whitelist?

3. No editor SQL do console do Data Management (DMS), crie o banco de dados mongo_test:

   use mongo_test;

4. Crie as coleções game_sales, game_dimension e platform_dimension e insira dados de exemplo:

   // Game sales table (status: 1 = active, 0 = logically deleted)
   db.game_sales.insert(
     [
   {sale_id:0,game_id:101,platform_id:1,"sale_date":"2024-01-01",units_sold:500,sale_amt:2500,status:1},
     ]
   );
   
   // Game dimension table
   db.game_dimension.insert(
     [
   {game_id:101,"game_name":"SpaceInvaders","release_date":"2023-06-15","developer":"DevCorp","publisher":"PubInc"},
   {game_id:102,"game_name":"PuzzleQuest","release_date":"2023-07-20","developer":"PuzzleDev","publisher":"QuestPub"},
   {game_id:103,"game_name":"RacingFever","release_date":"2023-08-10","developer":"SpeedCo","publisher":"RaceLtd"},
   {game_id:104,"game_name":"AdventureLand","release_date":"2023-09-05","developer":"Adventure","publisher":"LandCo"},
     ]
   );
   
   // Platform dimension table
   db.platform_dimension.insert(
     [
   {platform_id:1,"platform_name":"PCGaming","type":"PC"},
   {platform_id:2,"platform_name":"PlayStation","type":"Console"},
   {platform_id:3,"platform_name":"Mobile","type":"Mobile"}
     ]
   );

5. Verifique as inserções:

   db.game_sales.find();
   db.game_dimension.find();
   db.platform_dimension.find();

   A consulta db.game_dimension.find() retorna 4 registros:

   • game_id: 101, game_name: SpaceInvaders, release_date: 2023-06-15

   • game_id: 102, game_name: PuzzleQuest, release_date: 2023-07-20

   • game_id: 103, game_name: RacingFever, release_date: 2023-08-10

   • game_id: 104, game_name: AdventureLand, release_date: 2023-09-05

Criar a tabela do Hologres

1. Faça login no Hologres console, clique em Instances no painel de navegação à esquerda e selecione sua instância do Hologres. No canto superior direito, clique em Connect to Instance.

2. Na barra de navegação superior, clique em Metadata Management > Create Database. Insira test no campo Database Name, defina Policy como SPM e clique em OK. Para mais informações, consulte Create a database.

   Na caixa de diálogo, selecione a instância chamada User-behavior-test, defina Log On Immediately como Yes e clique em OK.

3. Na barra de navegação superior, clique em SQL Editor. Clique no ícone SQL para criar uma nova consulta, selecione a instância e o banco de dados de destino e execute a seguinte instrução para criar a tabela larga game_sales_details:

   CREATE TABLE game_sales_details(
     sale_id INT not null primary key,
     game_id INT,
     platform_id INT,
     sale_date VARCHAR(50),
     units_sold INT,
     sale_amt INT,
     status INT,
     game_name VARCHAR(50),
     release_date VARCHAR(50),
     developer VARCHAR(50),
     publisher VARCHAR(50),
     platform_name VARCHAR(50),
     type VARCHAR(50)
   );

Criar o tópico Kafka

1. Faça login no ApsaraMQ for Kafka console. Clique em Instances no painel de navegação à esquerda e selecione sua instância.

2. No painel de navegação à esquerda, clique em Whitelist Management e adicione o bloco CIDR do seu workspace Flink.

3. No painel de navegação à esquerda, clique em Topics > Create Topic. No painel direito, insira game_sales_fact no campo Name, adicione uma descrição e mantenha os valores padrão nos demais campos. Clique em OK.

Job 1: Gravar chaves primárias no Kafka

O Job 1 monitora as três coleções do MongoDB. Quando game_sales é alterado, sale_id é enviado diretamente para o Kafka. Se houver mudança em uma tabela de dimensões, um lookup join em game_sales recupera os valores de sale_id afetados, que são então transmitidos para o Kafka.

O conector do MongoDB desempenha duas funções neste pipeline. No Job 1, atua como source CDC, lendo o fluxo de alterações do MongoDB. No Job 2, funciona como source de lookup, buscando o estado atual de cada documento pela chave primária. Ambas as funções utilizam a mesma configuração de conector.

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

2. Na coluna Actions do seu workspace, clique em Console.

3. No menu de navegação à esquerda, clique em Development > ETL.

4. Clique em New Blank Stream Draft.

5. Na caixa de diálogo New Draft, insira dwd_mongo_kafka em Name, selecione uma versão do mecanismo e clique em Create.

6. Copie o SQL abaixo para o editor. Cada uma das três instruções INSERT captura independentemente as alterações de uma coleção do MongoDB e transmite os valores de sale_id afetados para o sink do Kafka. Isso garante que o Hologres seja atualizado com precisão e em tempo real sempre que qualquer tabela for modificada. Armazene valores sensíveis, como strings de conexão e senhas, como variáveis em vez de codificá-los diretamente. Para mais informações, consulte Manage variables.

   Tempo	**Estado de game_dimension para game_id = 101**
   T1	game_name = "SpaceInvaders"
   T2	Atualizado para game_name = "SpaceInvaders_v2"

   -- Source: game_sales (CDC stream)
   CREATE TEMPORARY TABLE game_sales
   (
     `_id`       STRING,    -- MongoDB auto-generated ID
     sale_id     INT,       -- Sales ID
     PRIMARY KEY (_id) NOT ENFORCED
   )
   WITH (
     'connector' = 'mongodb',
     'uri' = '${secret_values.MongoDB-URI}',
     'database' = 'mongo_test',
     'collection' = 'game_sales'
   );
   
   -- Source: game_dimension (CDC stream)
   CREATE TEMPORARY TABLE game_dimension
   (
     `_id`        STRING,
     game_id      INT,
     PRIMARY KEY (_id) NOT ENFORCED
   )
   WITH (
     'connector' = 'mongodb',
     'uri' = '${secret_values.MongoDB-URI}',
     'database' = 'mongo_test',
     'collection' = 'game_dimension'
   );
   
   -- Source: platform_dimension (CDC stream)
   CREATE TEMPORARY TABLE platform_dimension
   (
     `_id`         STRING,
     platform_id   INT,
     PRIMARY KEY (_id) NOT ENFORCED
   )
   WITH (
     'connector' = 'mongodb',
     'uri' = '${secret_values.MongoDB-URI}',
     'database' = 'mongo_test',
     'collection' = 'platform_dimension'
   );
   
   -- Lookup source: game_sales (used for dimension-change joins)
   CREATE TEMPORARY TABLE game_sales_dim
   (
     `_id`       STRING,
     sale_id     INT,
     game_id     INT,
     platform_id INT,
     PRIMARY KEY (_id) NOT ENFORCED
   )
   WITH (
     'connector' = 'mongodb',
     'uri' = '${secret_values.MongoDB-URI}',
     'database' = 'mongo_test',
     'collection' = 'game_sales'
   );
   
   -- Sink: Kafka topic that stores affected PKs
   CREATE TEMPORARY TABLE game_sales_fact (
     sale_id      INT,
     PRIMARY KEY (sale_id) NOT ENFORCED
   ) WITH (
     'connector' = 'upsert-kafka',
     'properties.bootstrap.servers' = '${secret_values.Kafka-hosts}',
     'topic' = 'game_sales_fact',
     'key.format' = 'json',
     'value.format' = 'json',
     'properties.enable.idempotence' = 'false'  -- Required when writing to ApsaraMQ for Kafka
   );
   
   BEGIN STATEMENT SET;
   
   -- Stream PKs from game_sales changes
   INSERT INTO game_sales_fact (sale_id)
   SELECT sale_id FROM game_sales;
   
   -- Stream PKs of game_sales rows affected by game_dimension changes
   INSERT INTO game_sales_fact (sale_id)
   SELECT gs.sale_id
   FROM game_dimension AS gd
   JOIN game_sales_dim FOR SYSTEM_TIME AS OF PROCTIME() AS gs
   ON gd.game_id = gs.game_id;
   
   -- Stream PKs of game_sales rows affected by platform_dimension changes
   INSERT INTO game_sales_fact (sale_id)
   SELECT gs.sale_id
   FROM platform_dimension AS pd
   JOIN game_sales_dim FOR SYSTEM_TIME AS OF PROCTIME() AS gs
   ON pd.platform_id = gs.platform_id;
   
   END;

   Sobre lookup joins A cláusula FOR SYSTEM_TIME AS OF PROCTIME() define um lookup join (um tipo de junção temporal). No momento em que uma linha do source é processada, a junção busca a linha correspondente da tabela de dimensões no MongoDB e congela esse snapshot para o resultado da junção. Se a tabela de dimensões for atualizada posteriormente, as linhas já processadas não serão afetadas. Por exemplo: Uma linha game_sales processada em T1 faz junção com "SpaceInvaders". Uma linha processada em T2 faz junção com "SpaceInvaders_v2". A condição de junção é gd.game_id = gs.game_id e pd.platform_id = gs.platform_id. Para mais informações, consulte JOIN statements for dimension tables e Choose Kafka, Upsert Kafka, or Kafka JSON catalog.

7. No canto superior direito, clique em Deploy. Na caixa de diálogo, clique em Confirm. Para mais informações, consulte Deploy a job.

Job 2: Reconstruir a tabela larga e fazer upsert no Hologres

O Job 2 consome valores de sale_id do tópico Kafka game_sales_fact, executa lookup joins nas tabelas de fatos e dimensões do MongoDB e faz o upsert das linhas resultantes da tabela larga no Hologres.

Siga as etapas em Job 1 para criar um novo rascunho chamado dws_kafka_mongo_holo e implantá-lo com o seguinte SQL:

-- Source: Kafka topic that provides affected PKs
CREATE TEMPORARY TABLE game_sales_fact
(
  sale_id  INT,
  PRIMARY KEY (sale_id) NOT ENFORCED
)
WITH (
  'connector' = 'upsert-kafka',
  'properties.bootstrap.servers' = '${secret_values.Kafka-hosts}',
  'topic' = 'game_sales_fact',
  'key.format' = 'json',
  'value.format' = 'json',
  'properties.group.id' = 'game_sales_fact',
  'properties.auto.offset.reset' = 'earliest'
);

-- Lookup source: game_sales fact table
CREATE TEMPORARY TABLE game_sales
(
  `_id`       STRING,
  sale_id     INT,
  game_id     INT,
  platform_id INT,
  sale_date   STRING,
  units_sold  INT,
  sale_amt    INT,
  status      INT,
  PRIMARY KEY (_id) NOT ENFORCED
)
WITH (
  'connector' = 'mongodb',
  'uri' = '${secret_values.MongoDB-URI}',
  'database' = 'mongo_test',
  'collection' = 'game_sales'
);

-- Lookup source: game_dimension
CREATE TEMPORARY TABLE game_dimension
(
  `_id`        STRING,
  game_id      INT,
  game_name    STRING,
  release_date STRING,
  developer    STRING,
  publisher    STRING,
  PRIMARY KEY (_id) NOT ENFORCED
)
WITH (
  'connector' = 'mongodb',
  'uri' = '${secret_values.MongoDB-URI}',
  'database' = 'mongo_test',
  'collection' = 'game_dimension'
);

-- Lookup source: platform_dimension
CREATE TEMPORARY TABLE platform_dimension
(
  `_id`          STRING,
  platform_id    INT,
  platform_name  STRING,
  type           STRING,
  PRIMARY KEY (_id) NOT ENFORCED
)
WITH (
  'connector' = 'mongodb',
  'uri' = '${secret_values.MongoDB-URI}',
  'database' = 'mongo_test',
  'collection' = 'platform_dimension'
);

-- Sink: Hologres wide table
CREATE TEMPORARY TABLE IF NOT EXISTS game_sales_details
(
  sale_id       INT,
  game_id       INT,
  platform_id   INT,
  sale_date     STRING,
  units_sold    INT,
  sale_amt      INT,
  status        INT,
  game_name     STRING,
  release_date  STRING,
  developer     STRING,
  publisher     STRING,
  platform_name STRING,
  type          STRING,
  PRIMARY KEY (sale_id) NOT ENFORCED
)
WITH (
  'connector' = 'hologres',
  'dbname' = 'test',
  'tablename' = 'public.game_sales_details',
  'username' = '${secret_values.AccessKeyID}',
  'password' = '${secret_values.AccessKeySecret}',
  'endpoint' = '${secret_values.Hologres-endpoint}',
  'sink.delete-strategy' = 'IGNORE_DELETE',       -- Insert or update only; never delete rows
  'sink.on-conflict-action' = 'INSERT_OR_UPDATE',  -- Enable partial column updates
  'sink.partial-insert.enabled' = 'true'
);

INSERT INTO game_sales_details (
  sale_id, game_id, platform_id, sale_date, units_sold, sale_amt, status,
  game_name, release_date, developer, publisher, platform_name, type
)
SELECT
  gsf.sale_id,
  gs.game_id,
  gs.platform_id,
  gs.sale_date,
  gs.units_sold,
  gs.sale_amt,
  gs.status,
  gd.game_name,
  gd.release_date,
  gd.developer,
  gd.publisher,
  pd.platform_name,
  pd.type
FROM game_sales_fact AS gsf
JOIN game_sales FOR SYSTEM_TIME AS OF PROCTIME() AS gs
  ON gsf.sale_id = gs.sale_id
JOIN game_dimension FOR SYSTEM_TIME AS OF PROCTIME() AS gd
  ON gs.game_id = gd.game_id
JOIN platform_dimension FOR SYSTEM_TIME AS OF PROCTIME() AS pd
  ON gs.platform_id = pd.platform_id;

Atualizações da tabela de fatos

1. Insira mais cinco linhas em game_sales:

   db.game_sales.insert(
     [
   {sale_id:1,game_id:101,platform_id:1,"sale_date":"2024-01-01",units_sold:500,sale_amt:2500,status:1},
   {sale_id:2,game_id:102,platform_id:2,"sale_date":"2024-08-02",units_sold:400,sale_amt:2000,status:1},
   {sale_id:3,game_id:103,platform_id:1,"sale_date":"2024-08-03",units_sold:300,sale_amt:1500,status:1},
   {sale_id:4,game_id:101,platform_id:3,"sale_date":"2024-08-04",units_sold:200,sale_amt:1000,status:1},
   {sale_id:5,game_id:104,platform_id:2,"sale_date":"2024-08-05",units_sold:100,sale_amt:3000,status:1}
     ]
   );

   Consulte game_sales_details no Hologres. Cinco novas linhas aparecerão.

   A tabela de resultados da consulta contém colunas game_name (como SpaceInvaders, PuzzleQuest, RacingFever e AdventureLand) e release_date além dos campos sincronizados do MongoDB, exibindo os detalhes associados aos jogos.

2. Atualize sale_date de 2024-01-01 para 2024-08-01:

   db.game_sales.updateMany({"sale_date": "2024-01-01"}, {$set: {"sale_date": "2024-08-01"}});

   Consulte game_sales_details. A coluna sale_date refletirá o novo valor.

3. Exclua logicamente a linha onde sale_id = 5 definindo status como 0:

   db.game_sales.updateMany({"sale_id": 5}, {$set: {"status": 0}});

   Consulte game_sales_details. A coluna status para sale_id = 5 mudará para 0.

Atualizações da tabela de dimensões

1. Adicione novos jogos e plataformas às tabelas de dimensões:

   // New games
   db.game_dimension.insert(
     [
   {game_id:105,"game_name":"HSHWK","release_date":"2024-08-20","developer":"GameSC","publisher":"GameSC"},
   {game_id:106,"game_name":"HPBUBG","release_date":"2018-01-01","developer":"BLUE","publisher":"KK"}
     ]
   );
   
   // New platforms
   db.platform_dimension.insert(
     [
   {platform_id:4,"platform_name":"Steam","type":"PC"},
   {platform_id:5,"platform_name":"Epic","type":"PC"}
     ]
   );

   Inserir dados apenas nas tabelas de dimensões não aciona a sincronização — o pipeline é impulsionado por alterações em game_sales. Insira registros de vendas correspondentes para disparar a atualização da tabela larga:

   db.game_sales.insert(
     [
   {sale_id:6,game_id:105,platform_id:4,"sale_date":"2024-09-01",units_sold:400,sale_amt:2000,status:1},
   {sale_id:7,game_id:106,platform_id:1,"sale_date":"2024-09-01",units_sold:300,sale_amt:1500,status:1}
     ]
   );

   Consulte game_sales_details. Duas novas linhas aparecerão com os dados enriquecidos das dimensões.

2. Atualize os dados de dimensão no MongoDB:

   // Update release date
   db.game_dimension.updateMany({"release_date": "2018-01-01"}, {$set: {"release_date": "2024-01-01"}});
   
   // Update platform type
   db.platform_dimension.updateMany({"type": "PC"}, {$set: {"type": "Switch"}});

   Os campos atualizados se propagarão para as linhas correspondentes no Hologres.