背景信息

SCD簡(jiǎn)介

Slowly Changing Dimension（SCD）即緩慢變化維，是隨著時(shí)間變化的維度。在數(shù)據(jù)倉庫中存儲(chǔ)和管理當(dāng)前和歷史的數(shù)據(jù)，就需要考慮如何處理緩慢變化維，因此SCD被認(rèn)為是跟蹤維度變化的關(guān)鍵ETL任務(wù)之一。

G-SCD概念和解決方案

SCD處理維度新值的三種方式不能覆蓋業(yè)務(wù)的實(shí)際場(chǎng)景，所以E-MapReduce根據(jù)業(yè)務(wù)實(shí)際數(shù)倉場(chǎng)景提出了G-SCD（Based-Granularity Slowly Changing Dimension），即基于固定粒度（或者業(yè)務(wù)快照）的緩慢變化維。G-SCD按照固定的時(shí)間粒度生成一份業(yè)務(wù)快照數(shù)據(jù)，其中時(shí)間粒度可以是天、小時(shí)或者分鐘等，同時(shí)支持按照時(shí)間粒度查詢對(duì)應(yīng)時(shí)間段的數(shù)據(jù)。

其中按T保留全量數(shù)據(jù)的解決方案如下圖所示。

G-SCD on Delta Lake方案如下圖所示。

前提條件

已創(chuàng)建集群，詳情請(qǐng)參見創(chuàng)建集群。

使用限制

• 需要保證Kafka內(nèi)同一個(gè)Partition內(nèi)的數(shù)據(jù)嚴(yán)格有序。
• 數(shù)據(jù)按Key分區(qū)，保證同一Key必須落到同一個(gè)Kafka的Partition。

操作流程

1. 步驟一：創(chuàng)建G-SCD表
   創(chuàng)建G-SCD表，按照要求配置需要的參數(shù)。
2. 步驟二：處理數(shù)據(jù)
   您可以根據(jù)業(yè)務(wù)數(shù)據(jù)的情況，選擇使用流式寫入或者批量寫入的方式進(jìn)行數(shù)據(jù)的處理。示例中通過兩次批量寫入代替流式寫入的方式模擬G-SCD on Delta Lake的數(shù)據(jù)處理。
3. 步驟三：驗(yàn)證數(shù)據(jù)寫入結(jié)果
   通過查詢語句，驗(yàn)證數(shù)據(jù)是否寫入成功。

步驟一：創(chuàng)建G-SCD表

CREATE TABLE target (id Int, body String, dt string)
USING delta
TBLPROPERTIES (
  "delta.gscdTypeTable" = "true", 
  "delta.gscdGranularity" = "1 day",
  "delta.gscdColumnFormat" = "yyyy-MM-dd",
  "delta.gscdColumn" = "dt"
);

步驟二：處理數(shù)據(jù)

您可以根據(jù)業(yè)務(wù)數(shù)據(jù)的情況，選擇使用流式寫入或者批量寫入的方式進(jìn)行數(shù)據(jù)的處理。

• 流式寫入

  CREATE TABLE IF NOT EXISTS gscd_kafka_table
  USING kafka
  OPTIONS(
    kafka.bootstrap.servers = 'localhost:9092',
    subscribe = 'xxxxxx'
  );
  
  CREATE SCAN gscd_stream ON gscd_kafka_table USING STREAM
  OPTIONS (
    `watermark.time` = 'floor(ts/1000)'  --- 定義源頭watermark時(shí)間表達(dá)式，單位為秒。
  );
  
  CREATE STREAM delta_job
  OPTIONS (
    triggerType = 'ProcessingTime',
    checkpointLocation = '/path/to/checkpoint'
  )
  MERGE INTO gscd_target_table AS target
  USING (
    SELECT *, from_unixtime(ts/1000, 'yyyy-MM-dd') AS dt FROM gscd_stream
  ) AS source
  ON source.id = target.id AND target.dt = source.dt
  WHEN MATCHED THEN update set *
  WHEN NOT MATCHED THEN insert *;

  重要 在上述SQL語句中，watermark的使用原理及注意事項(xiàng)如下：

  • 為了在流作業(yè)中自動(dòng)觸發(fā)Savepoint，需要在CREATE SCAN語句中指定watermark時(shí)間表達(dá)式。
  • watermark表示流作業(yè)源頭的時(shí)間值，單位為秒。
  • watermark時(shí)間會(huì)生成作為delta.gscdColumn字段的值，當(dāng)watermark時(shí)間達(dá)到delta.gscdGranularity邊界時(shí)（示例中定義的為1 day），會(huì)自動(dòng)觸發(fā)Savepoint。
  • watermark時(shí)間要求在同一個(gè)Partition內(nèi)遞增有序。
• 批量寫入

  一般場(chǎng)景下，通過流式寫入已經(jīng)可以滿足。但當(dāng)數(shù)據(jù)異常時(shí)，G-SCD on Delta Lake的方案同時(shí)提供了回滾Rollback的能力，并可以使用批量離線寫入修復(fù)數(shù)據(jù)。修復(fù)完成后，執(zhí)行Savepoint，永久保留當(dāng)前Version。

  MERGE INTO GSCD("2021-01-01") gscd_target_table
  USING gscd_source_table
  ON source.id = target.id  
  WHEN MATCHED THEN UPDATE SET body = source.body
  WHEN NOT MATCHED THEN INSERT(id, body) VALUES(source.id, source.body);

  在上述SQL語句中，GSCD ("2021-01-01")的語法表示要寫入的數(shù)據(jù)所屬的業(yè)務(wù)粒度值。

  重要 批量寫入不支持同一個(gè)作業(yè)寫入多個(gè)業(yè)務(wù)粒度數(shù)據(jù)。如果存在這種情況，需要提前進(jìn)行拆分。

為了幫助您快速使用G-SCD處理維度數(shù)據(jù)，本文給出詳細(xì)的示例，具體操作步驟如下。

1. 模擬源數(shù)據(jù)。

   CREATE TABLE s1 (id Int, body String) USING delta;
   CREATE TABLE s2 (id Int, body String) USING delta;
   INSERT INTO s1 VALUES (1, "addr_1_v1"), (2, "addr_2_v1"), (3, "addr_3_v1");
   INSERT INTO s2 VALUES (2, "addr_2_v2"), (4, "addr_1_v1");

2. 通過使用兩次批量寫入，代替流式寫入的方式模擬G-SCD on Delta Lake的數(shù)據(jù)處理。
   1. 第一次批量寫入，然后創(chuàng)建對(duì)應(yīng)時(shí)間粒度的Savepoint。

      -- 第一次批量寫入。
      MERGE INTO GSCD ("2021-01-01") target as target
      USING s1 as source
      ON source.id = target.id  
      WHEN MATCHED THEN UPDATE SET body = source.body
      WHEN NOT MATCHED THEN INSERT(id, body) VALUES(source.id, source.body);
      
      -- 創(chuàng)建2021-01-01的Savepoint。
      CREATE SAVEPOINT target GSCD('2021-01-01');

   2. 第二次批量寫入，然后創(chuàng)建對(duì)應(yīng)時(shí)間粒度的Savepoint。

      -- 第二次批量寫入。
      MERGE INTO GSCD ("2021-01-02") target as target
      USING s2 as source
      ON source.id = target.id  
      WHEN MATCHED THEN UPDATE SET body = source.body
      WHEN NOT MATCHED THEN INSERT(id, body) VALUES(source.id, source.body);
      
      -- 創(chuàng)建2021-01-02的Savepoint。
      CREATE SAVEPOINT target GSCD('2021-01-02');

步驟三：驗(yàn)證數(shù)據(jù)寫入結(jié)果

通過查詢語句，驗(yàn)證數(shù)據(jù)是否寫入成功。查詢?cè)?a title="" class="xref" href="#section-2ps-5pl-nqw">步驟二：處理數(shù)據(jù)示例中兩次批量寫入的數(shù)據(jù)，具體操作如下。

1. 執(zhí)行以下命令，查詢第一次批量寫入的數(shù)據(jù)。

   select id, body from target where dt = '2021-01-01';

   重要

   • 查詢數(shù)據(jù)時(shí)，可以使用正常的SQL語法。
   • 查詢數(shù)據(jù)時(shí)，必須指定gscdColumn字段作為查詢條件，并且必須為=表達(dá)式，例如dt = '2021-01-01'。
2. 執(zhí)行以下命令，查詢第二次批量寫入的數(shù)據(jù)。

   select id, body from target where dt = '2021-01-02';

   如果能夠查詢到寫入的數(shù)據(jù)，則表明數(shù)據(jù)寫入成功。執(zhí)行上述查詢命令后，返回結(jié)果如下圖所示。