我們?cè)u(píng)估一下這個(gè)方案，他的優(yōu)點(diǎn)是僅依賴 Spark + Delta 架構(gòu)簡(jiǎn)潔、沒(méi)有在線服務(wù)、列存，分析速度非?？?。優(yōu)化之后的 MERGE INTO 語(yǔ)法速度也夠快。

這個(gè)方案，業(yè)務(wù)上是一個(gè) Copy On Write 的一個(gè)方案，它只需要 copy 少量的文件，可以讓延遲做的相對(duì)低。理論上，在更新的數(shù)據(jù)跟現(xiàn)有的存量沒(méi)有很大重疊的話，可以把天級(jí)別的延遲做到小時(shí)級(jí)別的延遲，性能也是可以跟得上的。

這個(gè)方案在 Hive 倉(cāng)庫(kù)處理 upsert 數(shù)據(jù)的路上已經(jīng)前進(jìn)了一小步了。但小時(shí)級(jí)別的延遲畢竟不如實(shí)時(shí)更有效，因此這個(gè)方案最大的缺點(diǎn)在 Copy On Write 的 Merge  有一定的開(kāi)銷(xiāo)，延遲不能做的太低。

第一部分大概現(xiàn)有的方案就是這么多，同時(shí)還需要再?gòu)?qiáng)調(diào)一下，upsert 之所以如此重要，是因?yàn)樵跀?shù)據(jù)湖的方案中，upsert 是實(shí)現(xiàn)數(shù)據(jù)庫(kù)準(zhǔn)實(shí)時(shí)、實(shí)時(shí)入湖的一個(gè)關(guān)鍵技術(shù)點(diǎn)。

最后，F(xiàn)link + Iceberg 的 CDC 導(dǎo)入方案的優(yōu)點(diǎn)是什么？

對(duì)比之前的方案，Copy On Write 跟 Merge On Read 都有適用的場(chǎng)景，側(cè)重點(diǎn)不同。Copy On Write 在更新部分文件的場(chǎng)景中，當(dāng)只需要重寫(xiě)其中的一部分文件時(shí)是很高效的，產(chǎn)生的數(shù)據(jù)是純 append 的全量數(shù)據(jù)集，在用于數(shù)據(jù)分析的時(shí)候也是最快的，這是 Copy On Write 的優(yōu)勢(shì)。

另外一個(gè)是 Merge On Read，即將數(shù)據(jù)連同 CDC flag 直接 append 到 Iceberg 當(dāng)中，在 merge 的時(shí)候，把這些增量的數(shù)據(jù)按照一定的組織格式、一定高效的計(jì)算方式與全量的上一次數(shù)據(jù)進(jìn)行一次 merge。這樣的好處是支持近實(shí)時(shí)的導(dǎo)入和實(shí)時(shí)數(shù)據(jù)讀?。贿@套計(jì)算方案的 Flink SQL 原生支持 CDC 的攝入，不需要額外的業(yè)務(wù)字段設(shè)計(jì)。

Iceberg 是統(tǒng)一的數(shù)據(jù)湖存儲(chǔ)，支持多樣化的計(jì)算模型，也支持各種引擎（包括 Spark、Presto、hive）來(lái)進(jìn)行分析；產(chǎn)生的 file 都是純列存的，對(duì)于后面的分析是非?？斓模籌ceberg 作為數(shù)據(jù)湖基于 snapshot 的設(shè)計(jì)，支持增量讀?。籌ceberg 架構(gòu)足夠簡(jiǎn)潔，沒(méi)有在線服務(wù)節(jié)點(diǎn)，純 table format 的，這給了上游平臺(tái)方足夠的能力來(lái)定制自己的邏輯和服務(wù)化。

首先我們來(lái)了解一下在整個(gè)數(shù)據(jù)湖里面批量更新的兩個(gè)場(chǎng)景。

• 第一批量更新的這種場(chǎng)景，在這個(gè)場(chǎng)景中我們使用一個(gè) SQL 更新了成千上萬(wàn)行的數(shù)據(jù)，比如歐洲的 GDPR 策略，當(dāng)一個(gè)用戶注銷(xiāo)掉自己的賬戶之后，后臺(tái)的系統(tǒng)是必須將這個(gè)用戶所有相關(guān)的數(shù)據(jù)全部物理刪除。

• 第二個(gè)場(chǎng)景是我們需要將 date lake 中一些擁有共同特性的數(shù)據(jù)刪除掉，這個(gè)場(chǎng)景也是屬于批量更新的一個(gè)場(chǎng)景，在這個(gè)場(chǎng)景中刪除的條件可能是任意的條件，跟主鍵（Primary key）沒(méi)有任何關(guān)系，同時(shí)這個(gè)待更新的數(shù)據(jù)集是非常大，這種作業(yè)是一個(gè)長(zhǎng)耗時(shí)低頻次的作業(yè)。

另外是 CDC 寫(xiě)入的場(chǎng)景，對(duì)于對(duì) Flink 來(lái)說(shuō)，一般常用的有兩種場(chǎng)景，第一種場(chǎng)景是上游的 Binlog 能夠很快速的寫(xiě)到 data lake 中，然后供不同的分析引擎做分析使用； 第二種場(chǎng)景是使用 Flink 做一些聚合操作，輸出的流是 upsert 類型的數(shù)據(jù)流，也需要能夠?qū)崟r(shí)的寫(xiě)到數(shù)據(jù)湖或者是下游系統(tǒng)中去做分析。如下圖示例中 CDC 寫(xiě)入場(chǎng)景中的 SQL 語(yǔ)句，我們使用單條 SQL 更新一行數(shù)據(jù)，這種計(jì)算模式是一種流式增量的導(dǎo)入，而且屬于高頻的更新。

在了解了基本的概念，下面介紹 iceberg 中 insert、update、delete 操作的設(shè)計(jì)。

下圖示例的 SQL 中展示的表包含兩個(gè)字段即 id、data，兩個(gè)字段都是 int 類型。在一個(gè) transaction 中我們進(jìn)行了圖示中的數(shù)據(jù)流操作，首先插入了（1，2）一條記錄，接下來(lái)將這條記錄更新為（1，3），在 iceberg 中 update 操作將會(huì)拆為  delete 和 insert 兩個(gè)操作。

這么做的原因是考慮到 iceberg 作為流批統(tǒng)一的存儲(chǔ)層，將 update 操作拆解為 delete 和 insert 操作可以保證流批場(chǎng)景做更新時(shí)讀取路徑的統(tǒng)一，如在批量刪除的場(chǎng)景下以 Hive 為例，Hive 會(huì)將待刪除的行的文件 offset 寫(xiě)入到  delta 文件中，然后做一次 merge on read，因?yàn)檫@樣會(huì)比較快，在 merge 時(shí)通過(guò)  position 將原文件和 delta 進(jìn)行映射，將會(huì)很快得到所有未刪除的記錄。

接下來(lái)又插入記錄（3，5），刪除了記錄（1，3），插入記錄（2，5），最終查詢是我們得到記錄（3，5）（2，5）。

那么如何來(lái)解決這個(gè)問(wèn)題呢，社區(qū)當(dāng)前的方式是采用了 Mixed position-delete and equality-delete。Equality-delete 即通過(guò)指定一列或多列來(lái)進(jìn)行刪除操作，position-delete 是根據(jù)文件路徑和行號(hào)來(lái)進(jìn)行刪除操作，通過(guò)將這兩種方法結(jié)合起來(lái)以保證刪除操作的正確性。

如下圖我們?cè)诘谝粋€(gè) transaction 中插入了三行記錄，即 INSERT（1，2）、INSERT（1，3）、INSERT（1，4），然后執(zhí)行 commit 操作進(jìn)行提交。接下來(lái)我們開(kāi)啟一個(gè)新的 transaction 并執(zhí)行插入一行數(shù)據(jù)（1，5），由于是新的 transaction，因此新建了一個(gè) data file2 并寫(xiě)入 INSERT（1，5）記錄，接下來(lái)執(zhí)行刪除記錄（1，5)，實(shí)際寫(xiě)入 delete 時(shí)是：

1. 在 position delete file1 文件寫(xiě)入（file2, 0），表示刪除 data file2 中第 0 行的記錄，這是為了解決同一個(gè) transaction 內(nèi)同一行數(shù)據(jù)反復(fù)插入刪除的語(yǔ)義的問(wèn)題。

2. 在 equality delete file1 文件中寫(xiě)入 DELETE (1,5)，之所以寫(xiě)入這個(gè) delete  是為了確保本次 txn 之前寫(xiě)入的 (1,5) 能被正確刪除。

然后執(zhí)行刪除（1，4）操作，由于（1，4）在當(dāng)前 transaction 中未曾插入過(guò)，因此該操作會(huì)使用 equality-delete 操作，即在 equality delete file1 中寫(xiě)入（1，4）記錄。在上述流程中可以看出在當(dāng)前方案中存在 data file、position delete file、equality delete file 三類文件。

在了解了寫(xiě)入流程后，如何來(lái)讀取呢。如下圖所示，對(duì)于 position delete file 中的記錄（file2, 0）只需和當(dāng)前 transaction 的 data file 進(jìn)行 join 操作，對(duì)于 equality delete file 記錄（1，4）和之前的 transaction 中的 data file 進(jìn)行 join 操作。最終得到記錄 INSERT（1，3）、INSERT（1，2）保證了流程的正確性。

上面介紹了 insert、update 及 delete，但在設(shè)計(jì) task 的執(zhí)行計(jì)劃時(shí)我們對(duì) manifest 進(jìn)行了一些設(shè)計(jì)，目的是通過(guò) manifest 能夠快速到找到 data file，并按照數(shù)據(jù)大小進(jìn)行分割，保證每個(gè) task 處理的數(shù)據(jù)盡可能的均勻分布。

如下圖示例，包含四個(gè) transaction，前兩個(gè) transaction 是 INSERT 操作，對(duì)應(yīng) M1、M2，第三個(gè) transaction 是 DELETE 操作，對(duì)應(yīng) M3，第四個(gè) transaction 是 UPDATE 操作，包含兩個(gè) manifest 文件即 data manifest 和 delete manifest。

另一個(gè)問(wèn)題是我們需要保證足夠高的并發(fā)讀取，在 iceberg 中這點(diǎn)做得非常出色。在 iceberg 中可以做到文件級(jí)別的并發(fā)讀取，甚至文件中更細(xì)粒度的分段的并發(fā)讀取，比如文件有 256MB，可以分為兩個(gè) 128MB 進(jìn)行并發(fā)讀取。這里舉例說(shuō)明，假設(shè) insert 文件跟 delete 文件在兩個(gè) Bucket 中的布局方式如下圖所示。

前面介紹了文件的寫(xiě)入，下圖我們介紹如何按照 iceberg 的語(yǔ)義進(jìn)行寫(xiě)入并且供用戶讀取。主要分為數(shù)據(jù)和 metastore 兩部分，首先會(huì)有 IcebergStreamWriter 進(jìn)行數(shù)據(jù)的寫(xiě)入，但此時(shí)寫(xiě)入數(shù)據(jù)的元數(shù)據(jù)信息并沒(méi)有寫(xiě)入到 metastore，因此對(duì)外不可見(jiàn)。第二個(gè)算子是 IcebergFileCommitter，該算子會(huì)將數(shù)據(jù)文件進(jìn)行收集, 最終通過(guò) commit transaction 來(lái)完成寫(xiě)入。

在 Iceberg 中并沒(méi)有其他任何其他第三方服務(wù)的依賴，而 Hudi 在某些方面做了一些 service 的抽象，如將 metastore 抽象為獨(dú)立的 Timeline，這可能會(huì)依賴一些獨(dú)立的索引甚至是其他的外部服務(wù)來(lái)完成。

下面是我們未來(lái)的一些規(guī)劃，首先是 Iceberg 內(nèi)核的一些優(yōu)化，包括方案中涉及到的全鏈路穩(wěn)定性測(cè)試及性能的優(yōu)化， 并提供一些 CDC 增量拉取的相關(guān) Table API 接口。

在 Flink 集成上，會(huì)實(shí)現(xiàn) CDC 數(shù)據(jù)的自動(dòng)和手動(dòng)合并數(shù)據(jù)文件的能力，并提供  Flink 增量拉取 CDC 數(shù)據(jù)的能力。

在其他生態(tài)集成上，我們會(huì)對(duì) Spark、Presto 等引擎進(jìn)行集成，并借助 Alluxio 加速數(shù)據(jù)查詢。