Iceberg 實戰(zhàn) | Flink + Iceberg，百億級實時數據入湖實戰(zhàn)

摘要：本文整理自騰訊數據湖研發(fā)高級工程師陳俊杰在 4 月 17 日 上海站 Flink Meetup 分享的《百億級實時數據入湖實戰(zhàn)》。內容包括：

1. 騰訊數據湖介紹
   

2. 百億級數據場景落地
   

3. 未來規(guī)劃

4. 總結
   

   
   

Tips：點擊文末「閱讀原文」即可查看更多技術干貨～

 GitHub 地址 

歡迎大家給 Flink 點贊送 star~

一、騰訊數據湖介紹

從上圖可以看出來，整個平臺比較大，包括了數據接入、上層的分析、中間的管理 (如任務管理，分析管理和引擎管理)，再到最下層的 Table Format。

二、百億級數據落地場景落地

1. 傳統(tǒng)平臺架構

如上圖所示，過去的傳統(tǒng)平臺架構無非是兩種，一種是 Lambda 架構，一種是 Kappa 架構：

• Lambda 架構中，批和流是分開的，所以運維要有兩套集群，一套是 For Spark/Hive，一套是 For Flink。這存在幾個問題：

  
  

• 第一是運維的成本比較大；
• 第二是開發(fā)成本。例如在業(yè)務方面，一會要寫 Spark，一會要寫 Flink 或者 SQL，總體來說，開發(fā)成本對數據分析人員不是特別友好。
  
  

• 第二個是 Kappa 架構。其實就是消息隊列，到底層的傳輸，再到后面去做一些分析。它的特點是比較快，基于 Kafka 有一定的實時性。

這兩種架構各有利弊，最大的問題是存儲可能會不統(tǒng)一，導致數據鏈路割裂。目前我們平臺已經接入了 Iceberg，下面會根據不同場景，闡述遇到的問題及解決的過程。

2. 場景一: 手 Q 安全數據入湖

手機 QQ 安全數據入湖是一個非常典型的場景。

目前的業(yè)務場景是消息隊列 TubeMQ 通過 Flink 落地成 ODS 到 Iceberg，然后再用 Flink 做一些用戶表的關聯，之后做成一個寬表去做一些查詢，放到 COS 中，可能會在 BI 場景做一些分析。

這個過程看似平平無奇，但是要知道，手 Q 的用戶關聯維表為 28 億，每天的消息隊列是百億級的，因此會面臨一定的挑戰(zhàn)。

■ 小文件挑戰(zhàn)

1、Flink Writer 產生小文件

Flink 寫入沒有 shuffle，分發(fā)的數據無序，導致小文件多。

2、延遲要求高

checkpoint 間隔短，commit 間隔小，放大小文件問題。

3、小文件爆炸

幾天時間元數據和數據的小文件同時爆炸，集群壓力巨大。

4、合并小文件又放大問題

為了解決小文件問題，開 Action 進行小文件合并，結果產生更多文件。

5、來不及刪數據

刪除快照，刪孤兒文件，但是掃描文件太多，namenode 壓力巨大。

■  解決方案

1、Flink 同步合并

• 增加小文件合并 Operators；

  
  

• 增加 Snapshot 自動清理機制。

  
  

1）snapshot.retain-last.nums

2）snapshot.retain-last.minutes

2、Spark 異步合并

• 增加后臺服務進行小文件合并和孤兒文件刪除；

  
  

• 增加小文件過濾邏輯，逐步刪除小文件；

  
  

• 增加按分區(qū)合并邏輯，避免一次生成太多刪除文件導致任務 OOM。

■ Flink 同步合并

把所有的 Data 文件 Commit 之后，會產生一個 Commit Result。我們會拿 Commit Result 生成一個壓縮的任務，再給它并發(fā)成多個 Task Manager 去做 Rewrite 的工作，最終把結果 Commit 到 Iceberg 表里面。

當然，這里面的關鍵所在是 CompactTaskGenerator 怎么做。剛開始的時候我們想盡量地合并，于是去做表的 scan，把很多文件都掃一遍。然而它的表非常大，小文件非常多，一掃使得整個 Flink 立馬掛掉。

我們想了個方法，每次合并完，增量地去掃數據。從上一個 Replace Operation 里面到現在做一個增量，看這中間又增了多少，哪些符合 Rewrite 的策略。

這里面其實有許多配置，去看達到了多少個 snapshot，或者達到了多少個文件可以去做合并，這些地方用戶可以自己設置。當然，我們本身也設有默認值，從而保證用戶無感知地使用這些功能。

■ Fanout Writer 的坑

在 Fanout Writer 時，如果數據量大可能會遇到多層分區(qū)。比如手 Q 的數據分省、分市；但分完之后還是很大，于是又分 bucket。此時每個 Task Manager 里可能分到很多分區(qū)，每個分區(qū)打開一個 Writer，Writer 就會非常的多，造成內存不足。

這里我們做了兩件事情：

• 第一是 KeyBy 支持。根據用戶設置的分區(qū)做 KeyBy 的動作，然后把相同分區(qū)的聚集在一個 Task Manager 中，這樣它就不會打開那么多分區(qū)的 Writer。當然，這樣的做法會帶來一些性能上的損失。

  
  

• 第二是做 LRU Writer，在內存里面維持一個 Map。

  
  

3. 場景二：新聞平臺索引分析

上方是基于 Iceberg 流批一體的新聞文章在線索引架構。左邊是 Spark 采集 HDFS 上面的維表，右邊是接入系統(tǒng)，采集以后會用 Flink 和維表做一個基于 Window 的 Join，然后寫到索引流水表中。

■ 功能

• 準實時明細層；

  
  

• 實時流式消費；

  
  

• 流式 MERGE INTO；

  
  

• 多維分析；

  
  

• 離線分析。

■ 場景特點

上述場景有以下幾個特點：

• 數量級：索引單表超千億，單 batch 2000 萬，日均千億；

  
  

• 時延需求：端到端數據可見性分鐘級；

  
  

• 數據源：全量、準實時增量、消息流；

  
  

• 消費方式：流式消費、批加載、點查、行更新、多維分析。

■ 挑戰(zhàn)：MERGE INTO

有用戶提出了 Merge Into 的需求，因此我們從三個方面進行了思考：

• 功能：將每個 batch join 后的流水表 Merge into 到實時索引表，供下游使用；

• 性能：下游對索引時效性要求高，需要考慮 merge into 能追上上游的 batch 消費窗口；

• 易用性：Table API？還是 Action API？又或是 SQL API？

■ 解決方案

• 第一步

• 參考 Delta Lake 設計 JoinRowProcessor；

• 利用 Iceberg 的 WAP 機制寫臨時快照。

• 第二步

• 可選擇跳過 Cardinality-check；

• 寫入時可以選擇只 hash，不排序。

• 第三步

• 支持 DataframeAPI；

• Spark 2.4 支持 SQL；

• Spark 3.0 使用社區(qū)版本。

4. 場景三：廣告數據分析

■ 廣告數據主要有以下幾個特點：

• 數量級：日均千億 PB 數據，單條 2K；

  
  

• 數據源：SparkStreaming 增量入湖；

  
  

• 數據特點：標簽不停增加，schema 不停變換；

  
  

• 使用方式：交互式查詢分析。

■ 遇到的挑戰(zhàn)與對應的解決方案：

• 挑戰(zhàn)一：Schema 嵌套復雜，平鋪后近萬列，一寫就 OOM。

  
  

解決方案：默認每個 Parquet Page Size 設置為 1M，需要根據 Executor 內存進行 Page Size 設置。

• 挑戰(zhàn)二：30 天數據基本集群撐爆。

  解決方案：提供 Action 進行生命周期管理，文檔區(qū)分生命周期和數據生命周期。

• 挑戰(zhàn)三：交互式查詢。

  
  

  解決方案：

1）column projection；

2）predicate push down。

三、未來規(guī)劃

對于未來的規(guī)劃主要分為內核側與平臺側。

1. 內核側

在未來，我們希望在內核側有以下幾點規(guī)劃：

■ 更多的數據接入

• 增量入湖支持；

  
  

• V2 Format 支持；

  
  

• Row Identity 支持。

  
  

■ 更快的查詢

• 索引支持；

  
  

• Alloxio 加速層支持；

  
  

• MOR 優(yōu)化。

■ 更好的數據治理

• 數據治理 Action；

  
  

• SQL Extension 支持；

  
  

• 更好的元數據管理。

2、平臺側

在平臺側我們有以下幾點規(guī)劃：

■ 數據治理服務化

• 元數據清理服務化；

  
  

• 數據治理服務化。

■ 增量入湖支持

• Spark 消費 CDC 入湖；

  
  

• Flink 消費 CDC 入湖。

■ 指標監(jiān)控告警

• 寫入數據指標；

  
  

• 小文件監(jiān)控和告警。

四、總結

經過大量生產上的應用與實踐，我們得到三方面的總結：

• 可用性：通過多個業(yè)務線的實戰(zhàn)，確認 Iceberg 經得起日均百億，甚至千億的考驗。

  
  

• 易用性：使用門檻比較高，需要做更多的工作才能讓用戶使用起來。
  

• 場景支持：目前支持的入湖場景 還沒有 Hudi 多，增量讀取這塊也比較缺失，需要大家努力補齊。

  
  

---

另外～《Apache Flink-實時計算正當時》電子書重磅發(fā)布，本書將助您輕松 Get Apache Flink 1.13 版本最新特征，同時還包含知名廠商多場景 Flink 實戰(zhàn)經驗，學用一體，干貨多多！快掃描下方二維碼獲取吧～

（本次為搶鮮版，正式版將于 7 月初上線）

更多 Flink 相關技術交流，可掃碼加入社區(qū)釘釘大群～

▼ 關注「Flink 中文社區(qū)」，獲取更多技術干貨 ▼

---

  戳我，立即報名！