作業一：將銷售表主鍵寫入消息隊列

游戲銷售表將表主鍵（sale_id）存入消息隊列Kafka，流程如下圖：

作業二：同步更新主鍵信息到消息隊列

維度表與銷售表相互進行連接（維表JOIN）。若維度表或銷售表發生變更，將會把銷售表中受影響數據的主鍵（sale_id）更新至消息隊列Kafka。作業流程如下圖：

維表JOIN：通常用于使用從外部系統查詢的數據來豐富表。join 要求一個表具有處理時間屬性，還需要一個強制的相等連接條件，詳情請參見維表JOIN語句。在下列作業中，帶有process time屬性的FOR SYSTEM_TIME AS OF 子句在確保聯接處理game_sales表每一行時，都能與join條件匹配的維表行連接。它還防止連接的維表在未來發生更新時變更連接的結果。強制的連接條件則是gd.game_id = gsf.game_id和pd.platform_id = gsf.platform_id。

參考作業一新建并部署作業（MongoDB_joinTo_Kafka）。

//創建游戲維度表
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',
  'hosts' = '${secret_values.MongoDB-hosts}',
  'username' = '${secret_values.MongoDB-username}',
  'password' = '${secret_values.MongoDB-password}',
  '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',
  'hosts' = '${secret_values.MongoDB-hosts}',
  'username' = '${secret_values.MongoDB-username}',
  'password' = '${secret_values.MongoDB-password}',
  'database' = 'mongo_test',
  'collection' = 'platform_dimension'
);

//創建游戲銷售表
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,
  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 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'
);

BEGIN STATEMENT SET;

//游戲維度表與游戲銷售表關聯，有數據更新則將所影響的銷售表主鍵插入Kafka Topic
INSERT INTO game_sales_fact (
   sale_id
)
select
  gs.sale_id
from game_dimension as gd
join game_sales 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 FOR SYSTEM_TIME AS OF PROCTIME() as gs
on pd.platform_id = gs.platform_id;

END;

作業三：同步更新Hologres明細表

消費Kafka Topic當中的主鍵信息（sale_id），通過三個維表Join操作得到銷售明細信息，寫入最終的明細表。作業流程如下圖：

參考作業一新建并部署作業（MongoDB_joinKafkaTo_Holo）。下列代碼中使用Hologres連接器創建了結果表game_sales_details。

//創建Kafka Topic存儲主鍵的事實表，消費主鍵信息
CREATE TEMPORARY TABLE 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 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 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 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 IF NOT EXISTS game_sales_details
(
  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_details', --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_details (
  sale_id,
  game_id,
  platform_id,
  sale_date,
  units_sold,
  sale_amt,
  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,
  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;