基于Flink+MongoDB+Hologres的游戏行业用户行为分析

用户行为数据通常具有庞大的数据量,存储成本较高,且缺乏统一的格式,导致处理难度较大。常用的宽表模型虽查询效率高,但冗余度高、存储空间大、维护复杂,更新慢。本文基于Flink+MongoDB+Hologres更好地实现宽表的数据分析,以游戏行业的用户行为数据分析为示例,构建用户行为数据宽表进行数据分析的方案。

方案架构和优势

架构

实时计算Flink是强大的流式计算引擎,支持对海量实时数据高效处理。云数据库MongoDB是一种文档型的NoSQL数据库,具有数据存储结构灵活、读写性能高、支持复杂的查询条件等特点。实时数仓Hologres是一站式实时数仓,支持数据实时写入与更新,实时数据写入即可查。三者结合相辅相成,能够将复杂多变的数据进行联合计算处理,实现对大型数据即时更新、查询与分析。本文的示例方案架构如下:

  1. MongoDB的业务数据更新后,将数据主键通过消息队列Kafka采用Upsert的方式更新写入。

  2. 无论是事实表或维度表的更新,都会将所影响的数据主键通过Flink写入消息队列Kafka。

  3. Flink通过消费消息队列Kafka中的主键信息,根据主键获取MongoDB的完整的业务数据后更新写入到Hologres。

image

优势

该方案有如下优势:

  • 云数据库MongoDB版适用于高并发读写的场景的分片集群架构,针对高频写入的海量数据,可无限扩展性能及存储空间,解决写入效率低和存储空间不足的问题。

  • 无论是事实表还是维度表的更新,都会让结果表的最新数据进行更新。这一过程确保了数据更新的及时性,并且在处理海量数据时,仅对发生变更的数据进行更新,从而有效解决了更新效率低下的问题。

  • 实时数仓Hologres支持高效更新与修正,写入即可查,由其对外提供宽表模型的数据查询,有效提升数据查询与分析的效率。

实践场景

本文以某游戏厂商为例,实现了将用户在平台上购买游戏的行为数据进行实时预处理并写入Hologres即时查询的业务场景。

image
game_sales(游戏销售表)

sale_id(销售ID)

game_id(游戏ID)

platform_id(平台ID)

sale_date(销售日期)

units_sold(销售量)

sale_amt(销售金额)

status(逻辑字段)

game_dimension(游戏维度表)

game_id(游戏ID)

game_name(游戏名称)

release_date(发售日期)

developer(开发者)

publisher(发行商)

gameplatform_dimension(平台维度表)

platform_id(平台ID)

platform_name(平台名称)

type(终端类型)

game_sales_details(游戏销售明细表)

sale_id(销售ID)

game_id(游戏ID)

platform_id(平台ID)

sale_date(销售日期)

units_sold(销售量)

sale_amt(销售金额)

status(逻辑字段)

game_name(游戏名称)

release_date(发售日期)

developer(开发者)

publisher(发行商)

platform_name(平台名称)

type(终端类型)

方案阐述

在该实时宽表方案中,维表变更(如游戏信息、平台类型更新)能够实时同步到 Hologres 宽表,核心依赖四步链路:

  1. 捕获变更:Flink 实时监听 MongoDB 维表的数据变化。

  2. 定位影响:通过关联键(如 game_id)查找受影响的事实记录,提取主键(如 sale_id)。

  3. 触发重算:将主键写入 Kafka,通知下游作业“哪些数据需要刷新”。

  4. 增量更新:下游作业拉取最新数据,重组宽表,并通过 INSERT_OR_UPDATE 模式写入 Hologres。

前提条件

  • 已开通Flink工作空间,仅实时计算引擎VVR 8.0.5及以上版本支持该方案,详情请参见开通实时计算Flink

  • 已创建云数据库MongoDB,仅云数据库 MongoDB 4.0及以上版本支持该方案,详情请参见创建MongoDB分片集群实例

  • 已创建实时数仓Hologres,仅1.3及以上版本的独享Hologres实例支持该方案,详情请参见购买实时数仓Hologres

  • 已创建消息队列Kafka,详情请参见部署消息队列Kafka实例

  • 实时计算Flink版、云数据库 MongoDB、实时数仓Hologres和云消息队列Kafka需要在同一VPC下。如果不在同一VPC,需要先打通跨VPC的网络或者使用公网的形式访问,详情请参见如何访问跨VPC的其他服务?如何访问公网?

  • 通过RAM用户或RAM角色等身份访问对应资源时,需要其具备对应资源的权限。

步骤一:准备数据

  1. MongoDB创建数据库和三张业务表并插入数据。

    1. 登录实例

    2. Flink工作空间网段设置为白名单,详情请参见设置白名单如何设置白名单

    3. 数据管理DMS控制台SQL Console页面中,创建mongo_test数据库,命令如下:

      use mongo_test;
    4. mongo_test数据库创建game_sales(游戏销售表),game_dimension(游戏维度表)和platform_dimension(平台维度表)并插入数据,命令如下:

      //游戏销售表(status为逻辑字段,1为存在,0为删除)
      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},
        ]
      );
      
      //游戏维度表
      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"},
        ]
      );
      
      //平台维度表
      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. 查询创建的表信息,命令如下:

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

      image

  2. Hologres创建数据分析宽表。

    1. 登录Hologres控制台,在实例列表页面,单击目标实例,单击右上角的登录实例

    2. 单击上方导航栏的新建库,新建数据库名称为test,本示例权限策略选择SPM,详情请参见使用Hologres管理控制台创建数据库

      image

    3. 选择上方导航栏中的SQL编辑器,单击左侧导航栏上方的SQL图标,新建SQL查询,选择对应实例名和数据库,填入下方代码创建销售明细表。

      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)
      );
  3. 消息队列Kafka创建Topic。

    1. 登录Kafka控制台,在实例列表页面,单击目标实例名称。

    2. 单击左侧导航栏的白名单管理,添加或修改白名单分组,将Flink工作空间网段设置为白名单。

    3. 左侧导航栏选择Topic管理,单击创建Topic,名称为game_sales_fact,其余选择默认设置,单击确认

步骤二:创建流作业

作业一:将销售表主键写入消息队列

将销售表和维度表的更新都以销售表的主键(sale_id)写入消息队列Kafka。 销售表的更新直接写入主键(sale_id)。维度表的更新,通过与销售表进行维表JOIN后获取关联的销售表数据的对应主键(sale_id)写入消息队列Kafka。作业流程如下图:

image
  1. 登录实时计算控制台

  2. 单击目标工作空间操作列下的控制台

  3. 在左侧导航栏,单击数据开发 > ETL

  4. 单击新建后,在新建作业草稿对话框,选择空白的流作业草稿,单击下一步

  5. 修改文件名称dwd_mongo_kafka,单击创建

    下列代码中使用MongoDB连接器创建了源表game_sales,使用Upsert Kafka连接器创建了KafkaTopicgame_sales_fact。为避免作业中出现明文密码,造成安全隐患,可以参见变量管理来配置密码和地址等变量。

    三个INSERT语句分别捕获事实表自身、游戏维度表和平台维度表的变更,通过sale_id将变化实时传递至下游。该设计确保Hologres宽表实现“实时、准确、完整”更新。任一INSERT缺失都将导致对应场景变更遗漏,引发数据滞后或不一致。
    //创建游戏销售表
    CREATE TEMPORARY TABLE game_sales
    (
      `_id`       STRING,    --MongoDB自生成Id
      sale_id     INT,       --销售Id
      PRIMARY KEY (_id) NOT ENFORCED
    )
    WITH (
      'connector' = 'mongodb',      --使用的连接器
      'uri' = '${secret_values.MongoDB-URI}', --MongoDBURI
      'database' = 'mongo_test',    --数据库名称
      'collection' = 'game_sales'   --数据库表名
    );
    
    //创建游戏维度表
    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'
    );
    
    //创建平台维度表
    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'
    );
    
    // 创建游戏销售表(维表)
    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'
    );
    
    //创建Kafka Topic存储主键信息的事实表(结果表)
    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'  --使用阿里云消息队列Kafka,写入结果表需关闭此参数
      );
    
    BEGIN STATEMENT SET;
    
    // 插入游戏销售表主键
    INSERT INTO game_sales_fact (
      sale_id
    )
    SELECT
      sale_id
    FROM game_sales
    ;
    
    //游戏维度表与游戏销售表关联,有数据更新则将所影响的销售表主键插入Kafka Topic
    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;
    
    //平台维度表与销售表关联,有数据更新则将所影响的销售主键插入Kafka事实表
    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;
    说明
    • 本示例使用的是Upsert Kafka连接器,以Upsert方式向Kafka写入数据,其与Kafka连接器的区别请参见Kafka、Upsert KafkaKafka JSON catalog的选择

    • 维表JOIN:通常用于使用从外部系统查询的数据来丰富表。join 要求一个表具有处理时间属性,还需要一个强制的相等连接条件,详情请参见维表JOIN语句。在作业中,带有process time属性的FOR SYSTEM_TIME AS OF 子句在确保联接处理game_sales表每一行时,都能与join条件匹配的维表行连接。它还防止连接的维表在未来发生更新时变更连接的结果。强制的连接条件则是gd.game_id = gsf.game_idpd.platform_id = gsf.platform_id。

  6. 单击右上方的部署,单击确认

    更多部署参数详情请参见部署作业

作业二:消费主键信息同步更新Hologres明细表

消费Kafka Topic当中的主键信息(sale_id),通过三个维表Join操作得到销售明细信息,写入最终的明细表。作业流程如下图:

image

参考作业一新建并部署作业(dws_kafka_mongo_holo)。下列代码中使用Hologres连接器创建了结果表game_sales_details。

通过消费Kafka中的sale_id,基于主键实时关联三张维度表,利用FOR SYSTEM_TIME AS OF PROCTIME() 实现变更数据的动态捕获与快照关联,将最新维度信息与事实数据拼接后写入Hologres宽表。该JOIN逻辑确保了宽表能实时、完整地反映事实表与维度表的最新状态,支撑“实时、准确、完整”的数据融合与更新。
//创建Kafka Topic存储主键的事实表,消费主键信息
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'
);

//游戏销售表(维表)
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'
);


//游戏维度表(维表)
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'
);

//平台维度表(维表)
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'
);

//游戏销售明细表(结果表)
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 (
  // 下列参数采用了VVR11版本的参数配置
  'connector' = 'hologres',
  'dbname' = 'test', --Hologres的数据库名称
  'tablename' = 'public.game_sales_details', --Hologres用于接收数据的表名称
  'username' = '${secret_values.AccessKeyID}', --当前阿里云账号的AccessKey ID
  'password' = '${secret_values.AccessKeySecret}', --当前阿里云账号的AccessKey Secret
  'endpoint' = '${secret_values.Hologres-endpoint}', --当前Hologres实例VPC网络的Endpoint
  'sink.delete-strategy'='IGNORE_DELETE',       -- 撤回消息的处理策略,IGNORE_DELETE适用于仅需插入或更新数据,而无需删除数据的场景
  'sink.on-conflict-action'='INSERT_OR_UPDATE', -- 更新模式。宽表merge需要开启此参数,实现部分列更新。
  '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;

步骤三:启动作业

  1. 完成两个作业开发部署后,在左侧导航选择运维中心 > 作业运维启动两个作业。

  2. 作业运行后,前往Hologres控制台,对明细表game_sales_details进行查询。

    SELECT * FROM game_sales_details;

    此时game_sales_details表中插入了一条数据。

    image

步骤四:数据更新和查询

对于销售表和维度表的数据变更,能直接更新反馈到明细表上进行查询分析。下面将列举几种常见的数据变更操作,以展示Hologres宽表数据实时更新的情况。

销售表数据变更

  1. MongoDBgame_sales表插入五条数据,并观察Hologresgame_sale_details表的结果。

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

    查询Hologresgame_sale_details表,可以看到,该表增加了五条相应的数据。

    image

  2. MongoDBgame_salessale_date=2024-01-01的数据均修改为2024-08-01。

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

    查询Hologresgame_sale_details表,可以看到,该表中sale_date=2024-01-01的数据都被修改成2024-08-01了。

    image

  3. MongoDBgame_sales表进行逻辑删除,将sale_id=5的数据中的status字段改为0。

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

    查询Hologresgame_sale_details表,可以看到,该表中sale_id=5的数据中的status字段值被修改成0,实现了逻辑删除。

    image

维度表数据变更

  1. MongoDBgame_dimension表和platform_dimension表插入数据,插入新的游戏和新的平台数据。

    //游戏维度表
    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"}
      ]
    );
    
    //平台维度表
    db.platform_dimension.insert(
      [
    	{platform_id:4,"platform_name":"Steam","type":"PC"},
    	{platform_id:5,"platform_name":"Epic","type":"PC"}
      ]
    );

    新增游戏数据和新增平台数据,并不会增加明细表数据。用户需要产生购买或者下载行为才可以,所以继续添加game_sales表的数据。

    // 游戏销售表
    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}
      ]
    );

    而后查询Hologresgame_sale_details表,可以看到,明细表新增了两条关于新游戏和新平台的数据记录。

    image

  2. MongoDBgame_dimension表和platform_dimension表更新数据,更新游戏信息和平台信息。

    //更新开发日期
    db.game_dimension.updateMany({"release_date": "2018-01-01"}, {$set: {"release_date": "2024-01-01"}});
    
    //更新平台类型
    db.platform_dimension.updateMany({"type": "PC"}, {$set: {"type": "Swich"}});

    而后查询Hologresgame_sale_details表,可以看到相关信息的变更,开发日期和平台类型都被实时更新了。

    image

相关文档