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

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

1. 騰訊數(shù)據(jù)湖介紹
   

2. 百億級(jí)數(shù)據(jù)場景落地
   

3. 未來規(guī)劃

4. 總結(jié)
   

   
   

Tips：點(diǎn)擊文末「閱讀原文」即可查看更多技術(shù)干貨～

 GitHub 地址 

歡迎大家給 Flink 點(diǎn)贊送 star~

一、騰訊數(shù)據(jù)湖介紹

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

二、百億級(jí)數(shù)據(jù)落地場景落地

1. 傳統(tǒng)平臺(tái)架構(gòu)

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

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

  
  

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

• 第二個(gè)是 Kappa 架構(gòu)。其實(shí)就是消息隊(duì)列，到底層的傳輸，再到后面去做一些分析。它的特點(diǎn)是比較快，基于 Kafka 有一定的實(shí)時(shí)性。

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

2. 場景一: 手 Q 安全數(shù)據(jù)入湖

手機(jī) QQ 安全數(shù)據(jù)入湖是一個(gè)非常典型的場景。

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

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

■ 小文件挑戰(zhàn)

1、Flink Writer 產(chǎn)生小文件

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

2、延遲要求高

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

3、小文件爆炸

幾天時(shí)間元數(shù)據(jù)和數(shù)據(jù)的小文件同時(shí)爆炸，集群壓力巨大。

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

為了解決小文件問題，開 Action 進(jìn)行小文件合并，結(jié)果產(chǎn)生更多文件。

5、來不及刪數(shù)據(jù)

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

■  解決方案

1、Flink 同步合并

• 增加小文件合并 Operators；

  
  

• 增加 Snapshot 自動(dòng)清理機(jī)制。

  
  

1）snapshot.retain-last.nums

2）snapshot.retain-last.minutes

2、Spark 異步合并

• 增加后臺(tái)服務(wù)進(jìn)行小文件合并和孤兒文件刪除；

  
  

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

  
  

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

■ Flink 同步合并

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

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

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

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

■ Fanout Writer 的坑

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

這里我們做了兩件事情：

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

  
  

• 第二是做 LRU Writer，在內(nèi)存里面維持一個(gè) Map。

  
  

3. 場景二：新聞平臺(tái)索引分析

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

■ 功能

• 準(zhǔn)實(shí)時(shí)明細(xì)層；

  
  

• 實(shí)時(shí)流式消費(fèi)；

  
  

• 流式 MERGE INTO；

  
  

• 多維分析；

  
  

• 離線分析。

■ 場景特點(diǎn)

上述場景有以下幾個(gè)特點(diǎn)：

• 數(shù)量級(jí)：索引單表超千億，單 batch 2000 萬，日均千億；

  
  

• 時(shí)延需求：端到端數(shù)據(jù)可見性分鐘級(jí)；

  
  

• 數(shù)據(jù)源：全量、準(zhǔn)實(shí)時(shí)增量、消息流；

  
  

• 消費(fèi)方式：流式消費(fèi)、批加載、點(diǎn)查、行更新、多維分析。

■ 挑戰(zhàn)：MERGE INTO

有用戶提出了 Merge Into 的需求，因此我們從三個(gè)方面進(jìn)行了思考：

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

• 性能：下游對(duì)索引時(shí)效性要求高，需要考慮 merge into 能追上上游的 batch 消費(fèi)窗口；

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

■ 解決方案

• 第一步

• 參考 Delta Lake 設(shè)計(jì) JoinRowProcessor；

• 利用 Iceberg 的 WAP 機(jī)制寫臨時(shí)快照。

• 第二步

• 可選擇跳過 Cardinality-check；

• 寫入時(shí)可以選擇只 hash，不排序。

• 第三步

• 支持 DataframeAPI；

• Spark 2.4 支持 SQL；

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

4. 場景三：廣告數(shù)據(jù)分析

■ 廣告數(shù)據(jù)主要有以下幾個(gè)特點(diǎn)：

• 數(shù)量級(jí)：日均千億 PB 數(shù)據(jù)，單條 2K；

  
  

• 數(shù)據(jù)源：SparkStreaming 增量入湖；

  
  

• 數(shù)據(jù)特點(diǎn)：標(biāo)簽不停增加，schema 不停變換；

  
  

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

■ 遇到的挑戰(zhàn)與對(duì)應(yīng)的解決方案：

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

  
  

解決方案：默認(rèn)每個(gè) Parquet Page Size 設(shè)置為 1M，需要根據(jù) Executor 內(nèi)存進(jìn)行 Page Size 設(shè)置。

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

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

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

  
  

  解決方案：

1）column projection；

2）predicate push down。

三、未來規(guī)劃

對(duì)于未來的規(guī)劃主要分為內(nèi)核側(cè)與平臺(tái)側(cè)。

1. 內(nèi)核側(cè)

在未來，我們希望在內(nèi)核側(cè)有以下幾點(diǎn)規(guī)劃：

■ 更多的數(shù)據(jù)接入

• 增量入湖支持；

  
  

• V2 Format 支持；

  
  

• Row Identity 支持。

  
  

■ 更快的查詢

• 索引支持；

  
  

• Alloxio 加速層支持；

  
  

• MOR 優(yōu)化。

■ 更好的數(shù)據(jù)治理

• 數(shù)據(jù)治理 Action；

  
  

• SQL Extension 支持；

  
  

• 更好的元數(shù)據(jù)管理。

2、平臺(tái)側(cè)

在平臺(tái)側(cè)我們有以下幾點(diǎn)規(guī)劃：

■ 數(shù)據(jù)治理服務(wù)化

• 元數(shù)據(jù)清理服務(wù)化；

  
  

• 數(shù)據(jù)治理服務(wù)化。

■ 增量入湖支持

• Spark 消費(fèi) CDC 入湖；

  
  

• Flink 消費(fèi) CDC 入湖。

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

• 寫入數(shù)據(jù)指標(biāo)；

  
  

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

四、總結(jié)

經(jīng)過大量生產(chǎn)上的應(yīng)用與實(shí)踐，我們得到三方面的總結(jié)：

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

  
  

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

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

  
  

---

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

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

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

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

---

  戳我，立即報(bào)名！