Flink 和 Iceberg 如何解決資料入湖面臨的挑戰

GitHub 位址 https://github.com/apache/flink 歡迎大家給 Flink 點贊送 star~

一、資料入湖的核心挑戰

資料實時入湖可以分成三個部分，分别是資料源、資料管道和資料湖（數倉），本文的内容将圍繞這三部分展開。

1. Case #1：程式 BUG 導緻資料傳輸中斷

• 首先，當資料源通過資料管道傳到資料湖（數倉）時，很有可能會遇到作業有 BUG 的情況，導緻資料傳到一半，對業務造成影響；
• 第二個問題是當遇到這種情況的時候，如何重起作業，并保證資料不重複也不缺失，完整地同步到資料湖（數倉）中。

2. Case #2：資料變更太痛苦

• 資料變更

  當發生資料變更的情況時，會給整條鍊路帶來較大的壓力和挑戰。以下圖為例，原先是一個表定義了兩個字段，分别是 ID 和 NAME。此時，業務方面的同學表示需要将位址加上，以友善更好地挖掘使用者的價值。

  首先，我們需要把 Source 表加上一個列 Address，然後再把到 Kafka 中間的鍊路加上鍊，然後修改作業并重新開機。接着整條鍊路得一路改過去，添加新列，修改作業并重新開機，最後把資料湖（數倉）裡的所有資料全部更新，進而實作新增列。這個過程的操作不僅耗時，而且會引入一個問題，就是如何保證資料的隔離性，在變更的過程中不會對分析作業的讀取造成影響。

  

• 分區變更

  如下圖所示，數倉裡面的表是以 “月” 為機關進行分區，現在希望改成以 “天” 為機關做分區，這可能就需要将很多系統的資料全部更新一遍，然後再用新的政策進行分區，這個過程十分耗時。

  

3. Case #3：越來越慢的近實時報表？

當業務需要更加近實時的報表時，需要将資料的導入周期，從 “天” 改到 “小時”，甚至 “分鐘” 級别，這可能會帶來一系列問題。

如上圖所示，首先帶來的第一個問題是：檔案數以肉眼可見的速度增長，這将對外面的系統造成越來越大的壓力。壓力主要展現在兩個方面：

• 第一個壓力是，啟動分析作業越來越慢，Hive Metastore 面臨擴充難題，如下圖所示。

  

  • 随着小檔案越來越多，使用中心化的 Metastore 的瓶頸會越來越嚴重，這會造成啟動分析作業越來越慢，因為啟動作業的時候，會把所有的小檔案原資料都掃一遍。
  • 第二是因為 Metastore 是中心化的系統，很容易碰到 Metastore 擴充難題。例如 Hive，可能就要想辦法擴後面的 MySQL，造成較大的維護成本和開銷。

• 第二個壓力是掃描分析作業越來越慢。

  随着小檔案增加，在分析作業起來之後，會發現掃描的過程越來越慢。本質是因為小檔案大量增加，導緻掃描作業在很多個 Datanode 之間頻繁切換。

  

4. Case #4：實時地分析 CDC 資料很困難

大家調研 Hadoop 裡各種各樣的系統，發現整個鍊路需要跑得又快又好又穩定，并且有好的并發，這并不容易。

• 首先從源端來看，比如要将 MySQL 的資料同步到資料湖進行分析，可能會面臨一個問題，就是 MySQL 裡面有存量資料，後面如果不斷産生增量資料，如何完美地同步全量和增量資料到資料湖中，保證資料不多也不少。

  

• 此外，假設解決了源頭的全量跟增量切換，如果在同步過程中遇到異常，如上遊的 Schema 變更導緻作業中斷，如何保證 CDC 資料一行不少地同步到下遊。

  

• 整條鍊路的搭建，需要涉及源頭全量跟同步的切換，包括中間資料流的串通，還有寫入到資料湖（數倉）的流程，搭建整個鍊路需要寫很多代碼，開發門檻較高。

  

• 最後一個問題，也是關鍵的一個問題，就是我們發現在開源的生态和系統中，很難找到高效、高并發分析 CDC 這種變更性質的資料。

  

5. 資料入湖面臨的核心挑戰

• 資料同步任務中斷
  • 無法有效隔離寫入對分析的影響；
  • 同步任務不保證 exactly-once 語義。
• 端到端資料變更
  • DDL 導緻全鍊路更新更新複雜；
  • 修改湖/倉中存量資料困難。
• 越來越慢的近實時報表
  • 頻繁寫入産生大量小檔案；
  • Metadata 系統壓力大, 啟動作業慢；
  • 大量小檔案導緻資料掃描慢。
• 無法近實時分析 CDC 資料
  • 難以完成全量到增量同步的切換；
  • 涉及端到端的代碼開發，門檻高；
  • 開源界缺乏高效的存儲系統。

二、Apache Iceberg 介紹

1. Netflix：Hive 上雲痛點總結

Netflix 做 Iceberg 最關鍵的原因是想解決 Hive 上雲的痛點，痛點主要分為以下三個方面：

1.1 痛點一：資料變更和回溯困難

1. 不提供 ACID 語義。在發生資料改動時，很難隔離對分析任務的影響。典型操作如：INSERT OVERWRITE；修改資料分區；修改 Schema；
2. 無法處理多個資料改動，造成沖突問題；
3. 無法有效回溯曆史版本。

1.2 痛點二：替換 HDFS 為 S3 困難

1. 資料通路接口直接依賴 HDFS API；
2. 依賴 RENAME 接口的原子性，這在類似 S3 這樣的對象存儲上很難實作同樣的語義；
3. 大量依賴檔案目錄的 list 接口，這在對象存儲系統上很低效。

1.3 痛點三：太多細節問題

1. Schema 變更時，不同檔案格式行為不一緻。不同 FileFormat 甚至連資料類型的支援都不一緻；
2. Metastore 僅維護 partition 級别的統計資訊，造成不 task plan 開銷； Hive Metastore 難以擴充；
3. 非 partition 字段不能做 partition prune。

2. Apache Iceberg 核心特性

• 通用化标準設計
  • 完美解耦計算引擎
  • Schema 标準化
  • 開放的資料格式
  • 支援 Java 和 Python
• 完善的 Table 語義
  • Schema 定義與變更
  • 靈活的 Partition 政策
  • ACID 語義
  • Snapshot 語義
• 豐富的資料管理
  • 存儲的流批統一
  • 可擴充的 META 設計支援
  • 批更新和 CDC
  • 支援檔案加密
• 成本效益
  • 計算下推設計
  • 低成本的中繼資料管理
  • 向量化計算
  • 輕量級索引

3. Apache Iceberg File Layout

上方為一個标準的 Iceberg 的 TableFormat 結構，核心分為兩部分，一部分是 Data，一部分是 Metadata，無論哪部分都是維護在 S3 或者是 HDFS 之上的。

4. Apache Iceberg Snapshot View

上圖為 Iceberg 的寫入跟讀取的大緻流程。

可以看到這裡面分三層：

• 最上面黃色的是快照；
• 中間藍色的是 Manifest；
• 最下面是檔案。

每次寫入都會産生一批檔案，一個或多個 Manifest，還有快照。

比如第一次形成了快照 Snap-0，第二次形成快照 Snap-1，以此類推。但是在維護原資料的時候，都是增量一步一步做追加維護的。

這樣的話可以幫助使用者在一個統一的存儲上做批量的資料分析，也可以基于存儲之上去做快照之間的增量分析，這也是 Iceberg 在流跟批的讀寫上能夠做到一些支援的原因。

5. 選擇 Apache Iceberg 的公司

上圖為目前在使用 Apache Iceberg 的部分公司，國内的例子大家都較為熟悉，這裡大緻介紹一下國外公司的使用情況。

• NetFlix 現在是有數百PB的資料規模放到 Apache Iceberg 之上，Flink 每天的資料增量是上百T的資料規模。
• Adobe 每天的資料新增量規模為數T，資料總規模在幾十PB左右。
• AWS 把 Iceberg 作為資料湖的底座。
• Cloudera 基于 Iceberg 建構自己整個公有雲平台，像 Hadoop 這種 HDFS 私有化部署的趨勢在減弱，上雲的趨勢逐漸上升，Iceberg 在 Cloudera 資料架構上雲的階段中起到關鍵作用。
• 蘋果有兩個團隊在使用：
  • 一是整個 iCloud 資料平台基于 Iceberg 建構；
  • 二是人工智能語音服務 Siri，也是基于 Flink 跟 Iceberg 來建構整個資料庫的生态。

三、Flink 和 Iceberg 如何解決問題

回到最關鍵的内容，下面闡述 Flink 和 Iceberg 如何解決第一部分所遇到的一系列問題。

首先，同步鍊路用 Flink，可以保證 exactly once 的語義，當作業出現故障時，能夠做嚴格的恢複，保證資料的一緻性。

第二個是 Iceberg，它提供嚴謹的 ACID 語義，可以幫使用者輕松隔離寫入對分析任務的不利影響。

如上所示，當發生資料變更時，用 Flink 和 Iceberg 可以解決這個問題。

Flink 可以捕捉到上遊 Schema 變更的事件，然後把這個事件同步到下遊，同步之後下遊的 Flink 直接把資料往下轉發，轉發之後到存儲，Iceberg 可以瞬間把 Schema 給變更掉。

當做 Schema 這種 DDL 的時候，Iceberg 直接維護了多個版本的 Schema，然後老的資料源完全不動，新的資料寫新的 Schema，實作一鍵 Schema 隔離。

另外一個例子是分區變更的問題，Iceberg 做法如上圖所示。

之前按 “月” 做分區（上方黃色資料塊），如果希望改成按 “天” 做分區，可以直接一鍵把 Partition 變更，原來的資料不變，新的資料全部按 “天” 進行分區，語義做到 ACID 隔離。

第三個問題是小檔案對 Metastore 造成的壓力。

首先對于 Metastore 而言，Iceberg 是把原資料統一存到檔案系統裡，然後用 metadata 的方式維護。整個過程其實是去掉了中心化的 Metastore，隻依賴檔案系統擴充，是以擴充性較好。

另一個問題是小檔案越來越多，導緻資料掃描會越來越慢。在這個問題上，Flink 和 Iceberg 提供了一系列解決方案：

• 第一個方案是在寫入的時候優化小檔案的問題，按照 Bucket 來 Shuffle 方式寫入，因為 Shuffle 這個小檔案，寫入的檔案就自然而然的小。
• 第二個方案是批作業定期合并小檔案。
• 第三個方案相對智能，就是自動增量地合并小檔案。

4. Case #4：實時地分析CDC資料很困難

• 首先是是全量跟增量資料同步的問題，社群其實已有 Flink CDC Connected 方案，就是說 Connected 能夠自動做全量跟增量的無縫銜接。

• 第二個問題是在同步過程中，如何保證 Binlog 一行不少地同步到湖中， 即使中間碰到異常。

  對于這個問題，Flink 在 Engine 層面能夠很好地識别不同類型的事件，然後借助 Flink 的 exactly once 的語義，即使碰到故障，它也能自動做恢複跟處理。

• 第三個問題是搭建整條鍊路需要做不少代碼開發，門檻太高。

  在用了 Flink 和 Data Lake 方案後，隻需要寫一個 source 表和 sink 表，然後一條 INSERT INTO，整個鍊路就可以打通，無需寫任何業務代碼。

• 最後是存儲層面如何支援近實時的 CDC 資料分析。

四、社群 Roadmap

上圖為 Iceberg 的 Roadmap，可以看到 Iceberg 在 2019 年隻發了一個版本， 卻在 2020 年直接發了三個版本，并在 0.9.0 版本就成為頂級項目。

上圖為 Flink 與 Iceberg 的 Roadmap，可以分為 4 個階段。

• 第一個階段是 Flink 與 Iceberg 建立連接配接。
• 第二階段是 Iceberg 替換 Hive 場景。在這個場景下，有很多公司已經開始上線，落地自己的場景。
• 第三個階段是通過 Flink 與 Iceberg 解決更複雜的技術問題。
• 第四個階段是把這一套從單純的技術方案，到面向更完善的産品方案角度去做。

活動推薦

阿裡雲基于 Apache Flink 建構的企業級産品-實時計算Flink版現開啟活動：

99元試用

實時計算Flink版

(包年包月、10CU）即有機會獲得 Flink 獨家定制T恤；另包3個月及以上還有85折優惠！

了解活動詳情：

https://www.aliyun.com/product/bigdata/sc