Delta Lake CDF查詢返回的Schema是在原表的Schema基礎(chǔ)上追加以下三個額外字段：

• _change_type：引起變更的操作，取值如下：

  • insert：標(biāo)識數(shù)據(jù)為新插入的。

  • delete：標(biāo)識數(shù)據(jù)為剛刪除的。

  • update_preimage和update_postimage：標(biāo)識數(shù)據(jù)為更新，分別記錄其變更前的記錄和變更后的記錄。

• _commit_version：變更對應(yīng)的Delta表版本。

• _commit_timestamp：變更對應(yīng)的Delta表版本提交的時間。

EMR Spark2上使用Spark SQL語法需要額外配置以下參數(shù)，代碼如下所示。

spark-sql --conf spark.sql.externalTableValuedFunctions=table_changes

SQL語法如下所示。

-- Create Delta CDF-enabled Table
CREATE TABLE cdf_tbl (id int, name string, age int) USING delta
TBLPROPERTIES ("delta.enableChangeDataFeed" = "true");

-- Insert Into
INSERT INTO cdf_tbl VALUES (1, 'XUN', 32), (2, 'JING', 30);

-- Insert Overwrite
INSERT OVERWRITE TABLE cdf_tbl VALUES (1, 'a1', 30), (2, 'a2', 32), (3, "a3", 32);

-- Update
UPDATE cdf_tbl set age = age + 1;

-- Merge Into
CREATE TABLE merge_source (id int, name string, age int) USING delta;
INSERT INTO merge_source VALUES (1, "a1", 31), (2, "a2_new", 33), (4, "a4", 30);

MERGE INTO cdf_tbl target USING merge_source source
ON target.id = source.id
WHEN MATCHED AND target.id % 2 == 0 THEN UPDATE SET name = source.name
WHEN MATCHED THEN DELETE
WHEN NOT MATCHED THEN INSERT *;

-- Delete
DELETE FROM cdf_tbl WHERE age >= 32;

-- CDF Query
-- 查詢從版本0開始的所有的Change Data。
select * from table_changes("cdf_tbl", 0);
select * from table_changes("cdf_tbl", '2023-02-03 15:33:34'); --2023-02-03 15:33:34為commit0的提交時間戳。


-- 查詢版本4對應(yīng)的Change Data。
select * from table_changes("cdf_tbl", 4, 4);
select * from table_changes("cdf_tbl", '2023-02-03 15:34:06', '2023-02-03 15:34:06'); --2023-02-03 15:34:06為commit4的提交時間戳。