E-MapReduce:Delta Lake CDC を使用した増分データウェアハウスの構築

Delta Lake CDF によって返されるスキーマには、元のテーブルのスキーマと次のフィールドが含まれます。

• _change_type: データ変更の原因となった操作。有効な値：

  • insert: テーブルにデータを挿入します。

  • delete: テーブルからデータを削除します。

  • update_preimage および update_postimage: テーブルのデータを更新します。 update_preimage は変更前のデータを記録し、update_postimage は変更後のデータを記録します。

• _commit_version: データ変更が発生した Delta Lake テーブルのバージョンを指定します。

• _commit_timestamp: データ変更が発生した Delta Lake テーブルがコミットされた時刻を指定します。

EMR Spark 2 で Spark SQL 構文を使用するには、次のパラメーターを設定する必要があります。サンプルコード：

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

次のサンプルコードは、SQL 構文の使用方法の例を示しています。

-- Delta CDF 対応テーブルの作成
CREATE TABLE cdf_tbl (id int, name string, age int) USING delta
TBLPROPERTIES ("delta.enableChangeDataFeed" = "true");

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

-- 上書き挿入
INSERT OVERWRITE TABLE cdf_tbl VALUES (1, 'a1', 30), (2, 'a2', 32), (3, "a3", 32);

-- 更新
UPDATE cdf_tbl set age = age + 1;

-- マージ
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 FROM cdf_tbl WHERE age >= 32;

-- CDF クエリ
-- すべてのバージョンのテーブルから変更データをクエリします。データのクエリを開始するテーブルのバージョンは 0 です。
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 は、バージョン 0 のテーブルがコミットされた時刻です。


-- バージョン 4 のテーブルから変更データをクエリします。
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 は、バージョン 4 のテーブルがコミットされた時刻です。