Realtime Compute for Apache Flink:User behavior analysis with Flink, MongoDB, and Hologres

Change propagation flow

Each change flows through four stages:

1. Capture: Detect real-time changes from MongoDB dimension tables.

2. Propagate: When a dimension table changes, Job 1 performs a lookup join (for example, on game_id) to find affected rows in the fact table and extracts their PKs (for example, sale_id).

3. Trigger: Send the PKs to Kafka to notify Job 2 of pending refreshes.

4. Upsert: Job 2 fetches the latest data, reconstructs the wide table row, and upserts it into Hologres.

Create MongoDB collections

1. Log on to your ApsaraDB for MongoDB instance.

2. Add the CIDR block of your Flink workspace to the MongoDB whitelist. For details, see Configure a whitelist for an instance and How do I configure a whitelist?

3. In the SQL editor of the Data Management (DMS) console, create the mongo_test database:

   use mongo_test;

4. Create the game_sales, game_dimension, and platform_dimension collections and insert sample data:

   // 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. Verify the inserts:

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

   The db.game_dimension.find() query returns 4 records:

   • 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

Create the Hologres table

1. Log on to the Hologres console, click Instances in the left navigation pane, and then click your Hologres instance. In the upper-right corner, click Connect to Instance.

2. In the top navigation bar, click Metadata Management > Create Database. Enter test in the Database Name field, set Policy to SPM, and click OK. For more information, see Create a database.

   In the dialog box, select the instance named User-behavior-test, set Log On Immediately to Yes, and click OK.

3. In the top navigation bar, click SQL Editor. Click the SQL icon to create a new SQL query, select the target instance and database, and run the following statement to create the game_sales_details wide table:

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

Create the Kafka topic

1. Log on to the ApsaraMQ for Kafka console. Click Instances in the left navigation pane, and click your instance.

2. In the left navigation pane, click Whitelist Management and add the CIDR block of your Flink workspace.

3. In the left navigation pane, click Topics > Create Topic. In the right pane, enter game_sales_fact in the Name field, enter a description, and keep the default values for all other fields. Click OK.

Job 1: Write primary keys to Kafka

Job 1 monitors all three MongoDB collections. When game_sales changes, sale_id is streamed directly to Kafka. When a dimension table changes, a lookup join against game_sales retrieves the affected sale_id values, which are then streamed to Kafka.

The MongoDB connector serves two roles in this pipeline. In Job 1, it acts as a CDC source, reading the MongoDB change stream. In Job 2, it acts as a lookup source, fetching the current state of each document by primary key. Both roles use the same connector configuration.

1. Log on to the Realtime Compute for Apache Flink console.

2. In the Actions column of your workspace, click Console.

3. In the left navigation menu, click Development > ETL.

4. Click New Blank Stream Draft.

5. In the New Draft dialog, enter dwd_mongo_kafka in Name, select an engine version, and click Create.

6. Copy the following SQL into the editor. Each of the three INSERT statements independently captures changes from one MongoDB collection and streams the affected sale_id values to the Kafka sink. This ensures Hologres is updated accurately and in real time whenever any table changes. Store sensitive values such as connection strings and passwords as variables rather than hardcoding them. For more information, see Manage variables.

   Time	game_dimension state for game_id = 101
   T1	game_name = "SpaceInvaders"
   T2	Updated to 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;

   About lookup joins The FOR SYSTEM_TIME AS OF PROCTIME() clause defines a lookup join (a type of temporal join). At the moment a source row is processed, the join fetches the matching dimension table row from MongoDB and freezes that snapshot for the join result. If the dimension table is updated later, already-processed rows are not affected. For example: A game_sales row processed at T1 joins with "SpaceInvaders". A row processed at T2 joins with "SpaceInvaders_v2". The join condition is gd.game_id = gs.game_id and pd.platform_id = gs.platform_id. For more information, see JOIN statements for dimension tables and Choose Kafka, Upsert Kafka, or Kafka JSON catalog.

7. In the upper-right corner, click Deploy. In the dialog, click Confirm. For more information, see Deploy a job.

Job 2: Reconstruct the wide table and upsert to Hologres

Job 2 consumes sale_id values from the game_sales_fact Kafka topic, performs lookup joins against the MongoDB fact and dimension tables, and upserts the resulting wide-table rows into Hologres.

Follow the steps in Job 1 to create a new draft named dws_kafka_mongo_holo and deploy it with the following 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;

Fact table updates

1. Insert five more rows into 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}
     ]
   );

   Query game_sales_details in Hologres. Five new rows appear.

   The query result table contains game_name (such as SpaceInvaders, PuzzleQuest, RacingFever, and AdventureLand) and release_date columns in addition to the fields synchronized from MongoDB, showing the associated game details.

2. Update sale_date from 2024-01-01 to 2024-08-01:

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

   Query game_sales_details. The sale_date column reflects the new value.

3. Logically delete the row where sale_id = 5 by setting status to 0:

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

   Query game_sales_details. The status column for sale_id = 5 changes to 0.

Dimension table updates

1. Add new games and platforms to the dimension tables:

   // 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"}
     ]
   );

   Inserting into dimension tables alone does not trigger synchronization — the pipeline is driven by changes to game_sales. Insert corresponding sales records to trigger the wide table update:

   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}
     ]
   );

   Query game_sales_details. Two new rows appear with the enriched dimension data.

2. Update dimension data in 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": "Swich"}});

   The updated fields propagate to the corresponding rows in Hologres.