一 摘要

Apache Flink是目前大數據領域最流行的流批一體化計算引擎，而數據湖技術也是互聯網時代的產物，以Iceberg、Hudi和Delta為代表的數據湖技術應運而生。Iceberg目前已經提供對Apache Flink 1.11.x的集成支持，Flink可以通過DataStream API/Table API將數據寫入Iceberg。

二 背景及痛點

新浪和微博有強大的用戶群體，目前積累的數據已經達到幾百PB。微博的技術通常會采集應用App的埋點數據以及應用服務日志之類的數據，這些數據通過Kafka消息中間件接入數據倉庫。

2.1 數據平臺架構

在微博的數據倉庫中，有多種大數據存儲組件，譬如Hive/HBase/HDFS，計算引擎有MapReduce、Spark、Flink，根據用戶不同的需求，會應用不同的技術在大數據平臺中計算，將結果保存到MySQL、ES等支持快速查詢的關系型、非關系型數據庫中，接下來應用層就可以基于這些數據進行BI報表開發、用戶畫像，或者基于Presto這種OLAP工具進行統計查詢。

2.2 Lambda架構的痛點

在整個數據處理的過程中，我們會借用調用系統來調度應用程序，定期（T+1或者H+1）去執行一些Spark任務。離線數據處理的整個過程中存在著大量的數據延遲現象，這些數據可能是T+1輸出或者是H+1輸出。但是，業務方已經不再滿足于這些離線處理數據的方式，因此，我們也用Flink+Kafka去構建了實時流數據處理系統。

如下圖，就是原來使用的Lambda架構，Lambda架構將數倉分成離線層和實時層。也就是說，同一份數據會被處理兩次以上，同一套業務邏輯需要適配兩次開發。

例如在實時場景下計算PV、UV時，我們會用實時技術計算，這些數據指標會離開呈現出來。但是，我們有時候需要做趨勢分析，需要每天再重新計算一次PV、UV數據，比如在凌晨3點的時候，用Spark在調度系統上把前一天的數據全部重新再跑一遍。

很顯然，這兩個過程運行的時間不一致，跑的數據卻是完全相同的。重新跑一遍離線分析的數據，數據的更新成本很高，更嚴重的是二者的數據可能不一致（比如有延遲數據產生，離線數據比實時更準確）。

為了解決Lambda架構的痛點，就產生了Kappa架構，相信大家對這個架構也非常熟悉。

2.3 Kappa架構的痛點

Kappa架構解決了Lambda架構中離線和實時數據間不一致、運營和開發成本加班的問題，但是Kappa架構也有痛點。

首先，我們需要借用Kafka來構建實時場景，但是如果需要對ODS層數據做進一步的分析時，就要接入Flink計算引擎把數據寫入到DWD層的Kafka，同樣也會將一部分結果數據寫入到DWS層的Kafka。但是，如果想做簡單的數據分析時，又要將DWD和DWS層的數據寫入到ClickHouse、ES、MySQL或者是Hive里做進一步分析，這無疑帶來了鏈路的復雜性。

其次，Kappa架構是嚴重依賴于消息隊列的，我們知道消息隊列本身的準確性嚴格依賴它上游數據的順序，但是，消息隊列越多，發生亂序的可能性越大。通常情況下，ODS層的數據是絕對準確的，把ODS層數據經過計算之后寫入到DWD層時就會產生亂序，DWD到DWS更容易產生亂序，這樣的數據不一致性問題非常大。

那么有沒有一種架構，既能滿足實時性的需求，又能滿足離線計算的需求，同時還能減輕運營開發成本？解決Kappa架構的痛點呢？

2.4 痛點總結

2.5 實時數據倉庫建設需求

是否有一種技術，既能夠保證數據高效的回溯能力，支持數據更新，又能夠實現數據的流批讀寫，并且還能夠實現分鐘級別的數據接入。

這也是建設實時數據倉庫的迫切需要，實際上需要對Kappa架構進行改進升級，以解決Kappa架構中遇到的問題，接下來我們會進一步探討數據湖技術--Iceberg。

三 數據湖Apache Iceberg介紹

3.1 Iceberg是什么

官網對Iceberg的描述如下：

Apache Iceberg is an open table format for huge analytic datasets. Iceberg adds tables to Trino and Spark that use a high-performance format that works just like a SQL table.

Iceberg的官方定義是一種表格式，可以理解為是基于計算層（Spark、Flink）和存儲層（ORC、Parquet、Avro）的中間介質層，用Flink或者Spark將數據寫入Iceberg，然后通過Presto、Flink、Spark來讀取這些表。

3.2 Iceberg的Table格式介紹

Iceberg主要是為分析海量數據計算的，被定義為Table Format，Table Format介于計算層和存儲層之間。

Table Format向下管理存儲系統上的文件，向上為計算層提供接口。比如一張Hive表，在HDFS上會有Partition，存儲格式，壓縮格式和數據的HDFS目錄等，這些信息都維護在元數據中，這些元數據被稱為一種文件的組織形式。

Iceberg能夠高效支撐上層的計算層訪問磁盤上的文件。

3.3 Iceberg的功能總結

Iceberg目前支持三種文件格式，Parquet、ORC、Avro，Iceberg的主要功能如下：

3.4 Iceberg的設計

3.4.1 設計目標

• 和HIVE模式類似，它也是一種開放的靜態數據存儲形式，和計算層使用的語言不同。
• 具有強大的擴展性和可靠性：簡單透明的使用方式，用戶只需要關心寫入數據的邏輯，Iceberg會自動識別所有元數據的變更。Iceberg也支持并發寫。
• 存儲結構高可用：Iceberg有非常合理的Schema管理模式，具有多版本管理機制，支持版本回滾。

3.4.2 詳細設計

自帶ACID能力：保障每次寫入后的數據都是一個完整的快照（snapshot），每個snapshot包含著一系列的文件列表，落地任務把數據直接寫入Iceberg表中，不需要任務再做額外的success狀態維護。Iceberg會根據分區字段自動處理延時到來的數據，把延時的數據及時的寫入到正確的分區，因為有ACID的保障，延時數據寫入過程中Iceberg表依然提供可靠的讀取能力。

基于MVCC(Multi Version Concurrency Control)的機制，默認讀取文件會從最新的的版本，每次寫入都會產生一個新的snapshot，讀寫相互不干擾。

基于多版本的機制可以可用輕松實現回滾和時間旅行的功能，讀取或者回滾任意版本的snapshot數據。

3.4.3 組織架構

下圖是 Iceberg 整個文件的組織架構。從上往下看：

• 最上層是 snapshot 模塊。snapshot 是用戶可讀取的基本數據單位，也就是說，每次讀取一張表里面的所有數據，都是一個snapshot 里面的數據。
• 中間層manifest。一個 snapshot 下面會有多個 manifest，如圖 snapshot-0 有兩個 manifest，而 snapshot-1 有三個 manifest，每個 manifest 下面會管理一個或多個 DataFiles 文件。
• 數據層DataFiles。manifest 文件里面存放的就是數據的元信息，我們可以打開 manifest 文件，可以看到里面其實是一行行的 datafiles 文件路徑。

3.5 Iceberg的讀寫過程介紹

3.5.1 Iceberg的讀寫

如下圖所示，虛線框（snapshot-1）表示正在進行寫操作，但是還沒有發生commit操作，這時候 snapshot-1 是不可讀的，用戶只能讀取已經 commit 之后的 snapshot。同理， snapshot-2，snapshot-3表示已經可讀。

可以支持并發讀，例如可以同時讀取S1、S2、S3的快照數據，同時，可以回溯到snapshot-2或者snapshot-3。在snapshot-4 commit完成之后，這時候snapshot-4已經變成實線，就可以讀取數據了。

例如，現在current Snapshot 的指針移到S3，用戶對一張表的讀操作，都是讀 current Snapshot 指針所指向的 Snapshot，但不會影響前面的 snapshot 的讀操作。

3.5.2 增量讀取

Iceberg的每個snapshot都包含前一個snapshot的所有數據，每次都相當于全量讀取數據，對于整個鏈路來說，讀取數據的代價是非常高的。

如果我們只想讀取當前時刻的增量數據，就可以根據Iceberg中Snapshot的回溯機制來實現，僅讀取Snapshot1到Snapshot2的增量數據，也就是下圖中的紫色數據部分。

同理，S3也可以只讀取紅色部分的增量數據，也可以讀取S1-S3的增量數據。

Iceberg支持讀寫分離，也就是說可以支持并發讀和增量讀。

3.6 小文件問題

3.6.1 實時小文件問題

目前Flink社區現在已經重構了 Flink 里面的 FlinkIcebergSink，提供了 global committee 的功能，我們采用的也是社區提供的FlinkIcebergSink，曲線框中的這塊內容是 FlinkIcebergSink。

多個 IcebergStreamWriter 和一個 IcebergFileCommitter 的情況下，在上游的數據寫到 IcebergStreamWriter 的時候，每個 writer 里面做的事情都是去寫 datafiles 文件。

當每個 writer 寫完自己當前這一批 datafiles 小文件的時候，就會發送消息給 IcebergFileCommitter，告訴它可以提交了。而 IcebergFileCommitter 收到信息的時，就一次性將 datafiles 的文件提交，進行一次 commit 操作。

commit 操作本身只是對一些原始信息的修改，讓其從不可見變成可見。

3.6.2 實時合并小文件

在實際的生產環境中，Flink 實時作業會一直在集群中運行，為了要保證數據的時效性，一般會把 Iceberg commit 操作的時間周期設成 30 秒或者是一分鐘。當 Flink 作業跑一天時，如果是一分鐘一次 commit，一天需要 1440 個 commit，如果 Flink 作業跑一個月commit 操作會更多。甚至 snapshot commit 的時間間隔越短，生成的 snapshot 的數量會越多。當流式作業運行后，就會生成大量的小文件。

Iceberg 小文件合并是在
org.apache.iceberg.actions.RewriteDataFilesAction 類里面實現的。社區中小文件合并其實是通過 Spark 并行計算的，我們參考了社區Spark的實現方法，自己封裝了使用Flink合并小文件的方法。

四 Flink+Iceberg構建實時數倉

4.1 準實時數據倉庫分析系統

我們知道Iceberg支持讀寫分離，又支持并發讀、增量讀、合并小文件，而且還能做到秒級/分鐘級的數據延遲。我們基于Iceberg這些優勢，采用Flink+Iceberg的方式構建了流批一體化的實時數據倉庫。

在數據倉庫處理層，可以用 presto 進行一些簡單的查詢，因為 Iceberg 支持 Streaming read，所以在系統的中間層也可以直接接入 Flink，直接在中間層用 Flink 做一些批處理或者流式計算的任務，把中間結果做進一步計算后輸出到下游。

4.2 采用Iceberg替代Kafka實時數倉的優劣勢

五 未來規劃

5.1Iceberg 內核能力提升

• Row-level delete 功能。目前社區還不支持行級別的刪除功能，Iceberg 當前只支持 copy on write 的 update 的能力。如果要真正的構建一個實時數據倉庫，還是需要一個高效的 merge on read 的 update 能力。我們會繼續根據社區的更新動態，逐步迭代升級。
• 建立統一索引加速數據檢索。期待社區會有一個完善的統一索引加速功能。

5.2 內部大數據平臺升級

希望借助Alluxio構建一個數據湖加速功能，以便在查詢層實現秒級分析功能。

建立自動Schema建表的功能。

和所有業務系統打通，年內遷移完成所有業務線的數據，完整全部數據入湖建設。

原創不易，歡迎點贊加關注，您的關注是我持續創作的動力。