一、背景：

CDC數據中包含了，數據的變更過程。當CDC寫入傳統數據庫最終每一個primary key下會保存一條數據。當然可以使用特殊手段保存多分記錄但是顯然造成了數據膨脹。
另外數據湖Hudi(0.13.1)是不支持保存所有Changelog其Compaction機制會清除所有舊版本的內容。Iceberg支持TimeTravel，能查到某個時間點的數據狀態(tài)，但是不能列舉的單條記錄的Change過程。
所以目前只能手動實現。
其實，實現思路很簡單，將原PrimaryKey+Cdc的 ts_ms 一起作為新表的 PrimaryKey就可以了。但需要注意的是一條數據可能變更很多次，但一般需要保存近幾次的變更，所以就需要刪除部分舊變更記錄。ts_ms 就是CDC數據中記錄的日志實際產生的時間，具體參見debezium 。如果原表primarykey是聯合主鍵，即有多個字段共同組成，則最好將這些字段拼接為一個字符串，方便后續(xù)關聯。

本文思路
CDC --寫入-> Phoenix + 定期刪除舊版本記錄

CDC數據寫入略過，此處使用SQL模擬寫入。

二、Phoenix舊版記錄刪除（DEMO）

phoenix doc

bin/sqlline.py www.xx.com:2181

-- 直接創(chuàng)建phoenix表
create table TEST.TEST_VERSION(
ID VARCHAR NOT NULL,
TS TIMESTAMP NOT NULL,
NAME VARCHAR,
CONSTRAINT my_pk PRIMARY KEY (ID,TS)
) VERSIONS=5;

再去hbase shell中查看，hbase 關聯表已經有phoenix創(chuàng)建了。

hbase(main):032:0> desc "TEST:TEST_VERSION"
Table TEST:TEST_VERSION is ENABLED
TEST:TEST_VERSION, {TABLE_ATTRIBUTES => {coprocessor$1 => '|org.apache.phoenix.coprocessor.ScanRegionObserver|805306366|', coprocessor$2 => '|org.apache.phoenix.coprocessor.UngroupedAggregateRe
gionObserver|805306366|', coprocessor$3 => '|org.apache.phoenix.coprocessor.GroupedAggregateRegionObserver|805306366|', coprocessor$4 => '|org.apache.phoenix.coprocessor.ServerCachingEndpointImpl|80
5306366|', coprocessor$5 => '|org.apache.phoenix.hbase.index.Indexer|805306366|index.builder=org.apache.phoenix.index.PhoenixIndexBuilder,org.apache.hadoop.hbase.index.codec.class=org.apache.phoenix
.index.PhoenixIndexCodec', METADATA => {'OWNER' => 'dcetl'}}
COLUMN FAMILIES DESCRIPTION
{NAME => '0', VERSIONS => '5', EVICT_BLOCKS_ON_CLOSE => 'false', NEW_VERSION_BEHAVIOR => 'false', KEEP_DELETED_CELLS => 'FALSE', CACHE_DATA_ON_WRITE => 'false', DATA_BLOCK_ENCODING => 'FAST_DIFF', T
TL => 'FOREVER', MIN_VERSIONS => '0', REPLICATION_SCOPE => '0', BLOOMFILTER => 'NONE', CACHE_INDEX_ON_WRITE => 'false', IN_MEMORY => 'false', CACHE_BLOOMS_ON_WRITE => 'false', PREFETCH_BLOCKS_ON_OPE
N => 'false', COMPRESSION => 'NONE', BLOCKCACHE => 'true', BLOCKSIZE => '65536'}

-- 在phoenix中向表插入數據
UPSERT INTO TEST.TEST_VERSION(ID,TS,NAME) VALUES('rk001',TO_TIMESTAMP('2020-01-01 10:00:00'),'zhangsan');
UPSERT INTO TEST.TEST_VERSION(ID,TS,NAME) VALUES('rk001',TO_TIMESTAMP('2020-01-01 11:00:00'),'lisi');
UPSERT INTO TEST.TEST_VERSION(ID,TS,NAME) VALUES('rk001',TO_TIMESTAMP('2020-01-01 12:00:00'),'wangwu');
UPSERT INTO TEST.TEST_VERSION(ID,TS,NAME) VALUES('rk001',TO_TIMESTAMP('2020-01-01 13:00:00'),'zhaoliu');
UPSERT INTO TEST.TEST_VERSION(ID,TS,NAME) VALUES('rk001',TO_TIMESTAMP('2020-01-01 14:00:00'),'liuqi');
UPSERT INTO TEST.TEST_VERSION(ID,TS,NAME) VALUES('rk001',TO_TIMESTAMP('2020-01-01 15:00:00'),'sunba');
UPSERT INTO TEST.TEST_VERSION(ID,TS,NAME) VALUES('rk002',TO_TIMESTAMP('2020-01-01 07:00:00'),'sunyang');
UPSERT INTO TEST.TEST_VERSION(ID,TS,NAME) VALUES('rk002',TO_TIMESTAMP('2020-01-01 08:00:00'),'chaoyang');
UPSERT INTO TEST.TEST_VERSION(ID,TS,NAME) VALUES('rk002',TO_TIMESTAMP('2020-01-01 09:00:00'),'xuri');
UPSERT INTO TEST.TEST_VERSION(ID,TS,NAME) VALUES('rk002',TO_TIMESTAMP('2020-01-01 09:30:00'),'chenxi');
-- OK再查詢一下數據插入情況
SELECT * FROM TEST.TEST_VERSION;

以下假設每個PrimaryKey需要保留最新的3版本數據。所以紅色框內是需要刪除的數據。

現在需要使用row_number的函數給每個primarykey的不通version數據標識。但是phoenix并沒有開窗函數。只有agg聚合函數。
phoenix對SQL的限制還是比較多的如：
（1）join 非等值連接不支持，如on a.id>s.id 是不支持的，也不支持數組比較連接，如on a.id = ARRAY[1,2,3]。 會報錯：Error: Does not support non-standard or non-equi correlated-subquery conditions. (state=,code=0)
（2）where exists 格式的非等值連接不支持。select ... from A where exists (select 1 from B where A.id>B.id) 是不支持的。會報錯：Error: Does not support non-standard or non-equi correlated-subquery conditions. (state=,code=0)
（2）沒有開窗window函數
（3）DELETE FROM不支持JOIN

最終發(fā)下有一下函數可用
（1）NTH_VALUE 獲取分組排序的第N個值。 返回原值的類型。
（2）FIRST_VALUES 和 LAST_VALUES 獲取分區(qū)排序后的前、后的N個值，返回ARRAY類型。
此三個函數官網doc中，案例是這樣的 FIRST_VALUES( name, 3 ) WITHIN GROUP (ORDER BY salary DESC) 是全局分組，而實際使用中是需要搭配 GROUP BY 使用的。

所以可以獲取到

-- 方案一：使用NTH_VALUE獲取閾值
SELECT A.ID,A.TS FROM TEST.TEST_VERSION A 
INNER JOIN (
SELECT ID,NTH_VALUE(TS,3) WITHIN GROUP (ORDER BY TS DESC) THRES FROM TEST.TEST_VERSION GROUP BY ID) Z ON A.ID=Z.ID
WHERE A.TS < Z.THRES-- 方案二：使用FIRST_VALUES獲取到一個ARRAY 
SELECT A.ID,A.TS FROM TEST.TEST_VERSION A 
INNER JOIN (
SELECT ID,FIRST_VALUES(TS,3) WITHIN GROUP (ORDER BY TS DESC) TSS FROM TEST.TEST_VERSION GROUP BY ID) Z ON A.ID=Z.ID
WHERE A.TS < ALL(Z.TSS);

由于phoenix支持行子查詢，以下是官方案例。這樣就能繞過不使用DELETE … JOIN了。

Row subqueries
A subquery can return multiple fields in one row, which is considered returning a row constructor. The row constructor on both sides of the operator (IN/NOT IN, EXISTS/NOT EXISTS or comparison operator) must contain the same number of values, like in the below example:
SELECT column1, column2
FROM t1
WHERE (column1, column2) IN(SELECT column3, column4FROM t2WHERE column5 = ‘nowhere’);
This query returns all pairs of (column1, column2) that can match any pair of (column3, column4) in the second table after being filtered by condition: column5 = ‘nowhere’.

最終實現刪除 除N個較新的以外的所有舊版本數據， SQL如下：

-- NTH_VALUE方式
DELETE FROM TEST.TEST_VERSION
WHERE (ID,TS) IN (
SELECT A.ID,A.TS FROM TEST.TEST_VERSION A 
INNER JOIN (
SELECT ID,NTH_VALUE(TS,3) WITHIN GROUP (ORDER BY TS DESC) THRES FROM TEST.TEST_VERSION GROUP BY ID) Z ON A.ID=Z.ID
WHERE A.TS < Z.THRES
);-- FIRST_VALUES方式
DELETE FROM TEST.TEST_VERSION
WHERE (ID,TS) IN (
SELECT A.ID,A.TS FROM TEST.TEST_VERSION A 
INNER JOIN (
SELECT ID,FIRST_VALUES(TS,3) WITHIN GROUP (ORDER BY TS DESC) TSS FROM TEST.TEST_VERSION GROUP BY ID) Z ON A.ID=Z.ID
WHERE A.TS < ALL(Z.TSS)
);

刪除后效果：