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

作业一：销售表同步数据到Hologres和消息队列

游戏销售表的部分数据直接更新到销售事实表中，将表主键（sale_id）存入消息队列Kafka用于增量更新销售事实表，流程如下图：

作业二：同步更新主键信息到消息队列

维表与销售表相互进行连接（join）。若维表或销售表发生变更，将会把销售表中受影响数据的主键（sale_id）更新至消息队列Kafka。作业流程如下图：

参考作业一新建并部署作业。

//创建游戏维度表
CREATE TEMPORARY TABLE mongo_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',
  'hosts' = '${secret_values.MongoDB-hosts}',
  'username' = '${secret_values.MongoDB-username}',
  'password' = '${secret_values.MongoDB-password}',
  'database' = 'mongo_test',
  'collection' = 'game_dimension'
);

//创建平台维度表
CREATE TEMPORARY TABLE mongo_platform_dimension
(
  `_id`         STRING,
  platform_id   INT,
  platform_name STRING,
  type          STRING,
  PRIMARY KEY (_id) NOT ENFORCED
)
WITH (
  'connector' = 'mongodb',
  'hosts' = '${secret_values.MongoDB-hosts}',
  'username' = '${secret_values.MongoDB-username}',
  'password' = '${secret_values.MongoDB-password}',
  'database' = 'mongo_test',
  'collection' = 'platform_dimension'
);

//创建游戏销售表
CREATE TEMPORARY TABLE mongo_game_sales
(
  `_id`       STRING,
  sale_id     INT,
  game_id     INT,
  platform_id INT,
  sale_date   STRING,
  units_sold  INT,
  sale_amt    INT,
  PRIMARY KEY (_id) NOT ENFORCED
)
WITH (
  'connector' = 'mongodb',
  'hosts' = '${secret_values.MongoDB-hosts}',
  'username' = '${secret_values.MongoDB-username}',
  'password' = '${secret_values.MongoDB-password}',
  'database' = 'mongo_test',
  'collection' = 'game_sales'
);

//创建Kafka存储主键信息的事实表
CREATE TEMPORARY TABLE kafka_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'
);


BEGIN STATEMENT SET;

//游戏维度表与销售表关联，有数据更新则将所影响的销售主键插入Kafka事实表
INSERT INTO kafka_game_sales_fact (
   sale_id
)
select
  gsf.sale_id
from mongo_game_dimension as gd
inner join mongo_game_sales FOR SYSTEM_TIME AS OF PROCTIME() as gsf
on gd.game_id = gsf.game_id;

//平台维度表与销售表关联，有数据更新则将所影响的销售主键插入Kafka事实表
INSERT INTO kafka_game_sales_fact (
   sale_id
)
select
  gsf.sale_id
from mongo_platform_dimension as pd
inner join mongo_game_sales FOR SYSTEM_TIME AS OF PROCTIME() as gsf
on pd.platform_id = gsf.platform_id;

END;

作业三：同步更新Hologres宽表

通过消费Kafka当中的主键信息（sale_id），识别该主键的数据是新增或发生变更的数据，然后通过三张业务表相互join，将得到的数据更新写入到事实表中，作业流程如下图：

参考作业一新建并部署作业。

//创建Kafka存储主键的事实表，消费主键信息
CREATE TEMPORARY TABLE kafka_game_sales_fact (
  sale_id      INT,
  PRIMARY KEY (sale_id) NOT ENFORCED
) WITH (
  'connector' = 'kafka',
  'properties.bootstrap.servers' = '${secret_values.Kafka-hosts}',
  'topic' = 'game_sales_fact',    --消费的topic
  'format' = 'json',              --数据类型
  'properties.group.id' = 'game_sales_fact',  --消费者组
  'properties.auto.offset.reset' = 'earliest' --如果消费者组首次使用，则从最早位点开始消费
);


//游戏销售表
CREATE TEMPORARY TABLE mongo_game_sales
(
  `_id`       STRING,
  sale_id     INT,
  game_id     INT,
  platform_id INT,
  sale_date   STRING,
  units_sold  INT,
  sale_amt    INT,
  PRIMARY KEY (_id) NOT ENFORCED
)
WITH (
  'connector' = 'mongodb',
  'hosts' = '${secret_values.MongoDB-hosts}',
  'username' = '${secret_values.MongoDB-username}',
  'password' = '${secret_values.MongoDB-password}',
  'database' = 'mongo_test',
  'collection' = 'game_sales'
);

//游戏维度表
CREATE TEMPORARY TABLE mongo_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',
  'hosts' = '${secret_values.MongoDB-hosts}',
  'username' = '${secret_values.MongoDB-username}',
  'password' = '${secret_values.MongoDB-password}',
  'database' = 'mongo_test',
  'collection' = 'game_dimension'
);


//平台维度表
CREATE TEMPORARY TABLE mongo_platform_dimension
(
  `_id`          STRING
  ,platform_id   INT
  ,platform_name STRING
  ,type          STRING
  ,PRIMARY KEY (_id) NOT ENFORCED
)
WITH (
  'connector' = 'mongodb',
  'hosts' = '${secret_values.MongoDB-hosts}',
  'username' = '${secret_values.MongoDB-username}',
  'password' = '${secret_values.MongoDB-password}',
  'database' = 'mongo_test',
  'collection' = 'platform_dimension'
)
;

//Hologres结果表
CREATE TEMPORARY TABLE IF NOT EXISTS game_sales_fact
(
  sale_id       INT,
  game_id       INT,
  platform_id   INT,
  sale_date     STRING,
  units_sold    INT,
  sale_amt      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', --Hologres的数据库名称
  'tablename' = 'public.game_sales_fact', --Hologres用于接收数据的表名称
  'username' = '${secret_values.AccessKeyID}', --当前阿里云账号的AccessKey ID
  'password' = '${secret_values.AccessKeySecret}', --当前阿里云账号的AccessKey Secret
  'endpoint' = '${secret_values.Hologres-endpoint}', --当前Hologres实例VPC网络的Endpoint
  'ignoredelete' = 'false', -- 是否忽略撤回消息。宽表merge需要关闭。
  'mutatetype' = 'insertorupdate', -- 更新模式。宽表merge需要开启此参数，实现部分列更新。
  'partial-insert.enabled' = 'true', -- 是否只插入INSERT语句中定义的字段。宽表merge需要开启此参数，实现部分列更新。
  'ignoreNullWhenUpdate' = 'true' --是否忽略更新写入数据中的Null值。
);


INSERT INTO game_sales_fact (
  sale_id,
  game_id,
  platform_id,
  sale_date,
  units_sold,
  sale_amt,
  game_name,
  release_date,
  developer,
  publisher,
  platform_name,
  type
)
select
  kgsf.sale_id,
  gs.game_id,
  gs.platform_id,
  gs.sale_date,
  gs.units_sold,
  gs.sale_amt,
  gd.game_name,
  gd.release_date,
  gd.developer,
  gd.publisher,
  pd.platform_name,
  pd.type
from kafka_game_sales_fact as kgsf
inner join mongo_game_sales FOR SYSTEM_TIME AS OF PROCTIME() as gs
on kgsf.sale_id = gs.sale_id

inner join mongo_game_dimension FOR SYSTEM_TIME AS OF PROCTIME() as gd
on gs.game_id = gd.game_id

inner join mongo_platform_dimension FOR SYSTEM_TIME AS OF PROCTIME() as pd
on gs.platform_id = pd.platform_id;