Flink+Iceberg數(shù)據(jù)湖探索與實踐是怎樣的

這期內(nèi)容當中小編將會給大家?guī)碛嘘P(guān)Flink+Iceberg 數(shù)據(jù)湖探索與實踐是怎樣的，文章內(nèi)容豐富且以專業(yè)的角度為大家分析和敘述，閱讀完這篇文章希望大家可以有所收獲。

成都創(chuàng)新互聯(lián)是一家集網(wǎng)站建設(shè),濱江企業(yè)網(wǎng)站建設(shè),濱江品牌網(wǎng)站建設(shè),網(wǎng)站定制,濱江網(wǎng)站建設(shè)報價,網(wǎng)絡(luò)營銷,網(wǎng)絡(luò)優(yōu)化,濱江網(wǎng)站推廣為一體的創(chuàng)新建站企業(yè)，幫助傳統(tǒng)企業(yè)提升企業(yè)形象加強企業(yè)競爭力。可充分滿足這一群體相比中小企業(yè)更為豐富、高端、多元的互聯(lián)網(wǎng)需求。同時我們時刻保持專業(yè)、時尚、前沿，時刻以成就客戶成長自我，堅持不斷學(xué)習(xí)、思考、沉淀、凈化自己，讓我們?yōu)楦嗟钠髽I(yè)打造出實用型網(wǎng)站。

01 數(shù)據(jù)倉庫平臺建設(shè)的痛點

痛點一：

我們凌晨一些大的離線任務(wù)經(jīng)常會因為一些原因出現(xiàn)延遲，這種延遲會導(dǎo)致核心報表的產(chǎn)出時間不穩(wěn)定，有些時候會產(chǎn)出比較早，但是有時候就可能會產(chǎn)出比較晚，業(yè)務(wù)很難接受。

為什么會出現(xiàn)這種現(xiàn)象的發(fā)生呢？目前來看大致有這么幾點要素：

• 任務(wù)本身要請求的數(shù)據(jù)量會特別大。通常來說一天原始的數(shù)據(jù)量可能在幾十TB。幾百個分區(qū)，甚至上千個分區(qū)，五萬+的文件數(shù)這樣子。如果說全量讀取這些文件的話，幾百個分區(qū)就會向 NameNode 發(fā)送幾百次請求，我們知道離線任務(wù)在凌晨運行的時候，NameNode 的壓力是非常大的。所以就很有可能出現(xiàn) Namenode 響應(yīng)很慢的情況，如果請求響應(yīng)很慢就會導(dǎo)致任務(wù)初始化時間很長。

• 任務(wù)本身的 ETL 效率是相對低效的，這個低效并不是說 Spark 引擎低效，而是說我們的存儲在這塊支持的不是特別的好。比如目前我們查一個分區(qū)的話是需要將所有文件都掃描一遍然后進行分析，而實際上我可能只對某些文件感興趣。所以相對而言這個方案本身來說就是相對低效的。

• 這種大的離線任務(wù)一旦遇到磁盤壞盤或者機器宕機，就需要重試，重試一次需要耗費很長的時間比如幾十分鐘。如果說重試一兩次的話這個延遲就會比較大了。

痛點二：

針對一些細瑣的一些問題而言的。這里簡單列舉了三個場景來分析：

• 不可靠的更新操作。我們經(jīng)常在 ETL 過程中執(zhí)行一些 insert overwrite 之類的操作，這類操作會先把相應(yīng)分區(qū)的數(shù)據(jù)刪除，再把生成的文件加載到分區(qū)中去。在我們移除文件的時候，很多正在讀取這些文件的任務(wù)就會發(fā)生異常，這就是不可靠的更新操作。

• 表 Schema 變更低效。目前我們在對表做一些加字段、更改分區(qū)的操作其實是非常低效的操作，我們需要把所有的原始數(shù)據(jù)讀出來，然后在重新寫回去。這樣就會非常耗時，并且低效。

• 數(shù)據(jù)可靠性缺乏保障。主要是我們對于分區(qū)的操作，我們會把分區(qū)的信息分為兩個地方，HDFS 和 Metastore，分別存儲一份。在這種情況下，如果進行更新操作，就可能會出現(xiàn)一個更新成功而另一個更新失敗，會導(dǎo)致數(shù)據(jù)不可靠。

痛點三：

基于 Lambda 架構(gòu)建設(shè)的實時數(shù)倉存在較多的問題。如上圖的這個架構(gòu)圖，第一條鏈路是基于 kafka 中轉(zhuǎn)的一條實時鏈路（延遲要求小于5分鐘），另一條是離線鏈路（延遲大于1小時），甚至有些公司會有第三條準實時鏈路（延遲要求5分鐘～一小時），甚至更復(fù)雜的場景。

• 兩條鏈路對應(yīng)兩份數(shù)據(jù)，很多時候?qū)崟r鏈路的處理結(jié)果和離線鏈路的處理結(jié)果對不上。

• Kafka 無法存儲海量數(shù)據(jù)， 無法基于當前的 OLAP 分析引擎高效查詢 Kafka 中的數(shù)據(jù)。

• Lambda 維護成本高。代碼、數(shù)據(jù)血緣、Schema 等都需要兩套。運維、監(jiān)控等成本都非常高。

痛點四：

不能友好地支持高效更新場景。大數(shù)據(jù)的更新場景一般有兩種，一種是 CDC ( Change Data Capture) 的更新，尤其在電商的場景下，將 binlog 中的更新刪除同步到 HDFS 上。另一種是延遲數(shù)據(jù)帶來的聚合后結(jié)果的更新。目前 HDFS 只支持追加寫，不支持更新。因此業(yè)界很多公司引入了 Kudu。但是 Kudu 本身是有一些局限的，比如計算存儲沒有做到分離。這樣整個數(shù)倉系統(tǒng)中引入了 HDFS、Kafka 以及 Kudu，運維成本不可謂不大。

上面就是針對目前數(shù)倉所涉及到的四個痛點的大致介紹，因此我們也是通過對數(shù)據(jù)湖的調(diào)研和實踐，希望能在這四個方面對數(shù)倉建設(shè)有所幫助。接下來重點講解下對數(shù)據(jù)湖的一些思考。

02 數(shù)據(jù)湖 Iceberg 核心原理

1. 數(shù)據(jù)湖開源產(chǎn)品調(diào)研

數(shù)據(jù)湖大致是從19年開始慢慢火起來的，目前市面上核心的數(shù)據(jù)湖開源產(chǎn)品大致有這么幾個：

• DELTA LAKE，在17年的時候 DataBricks 就做了 DELTA LAKE 的商業(yè)版。主要想解決的也是基于 Lambda 架構(gòu)帶來的存儲問題，它的初衷是希望通過一種存儲來把 Lambda 架構(gòu)做成 kappa 架構(gòu)。

• Hudi ( Uber 開源 ) 可以支持快速的更新以及增量的拉取操作。這是它最大的賣點之一。

• Iceberg 的初衷是想做標準的 Table Format 以及高效的 ETL。

上圖是來自 Flink 團體針對數(shù)據(jù)湖方案的一些調(diào)研對比，總體來看這些方案的基礎(chǔ)功能相對都還是比較完善的。我說的基礎(chǔ)功能主要包括：

• 高效 Table Schema 的變更，比如針對增減分區(qū)，增減字段等功能

• ACID 語義保證

• 同時支持流批讀寫，不會出現(xiàn)臟讀等現(xiàn)象

• 支持 OSS 這類廉價存儲

2. 當然還有一些不同點：

Hudi 的特性主要是支持快速的更新刪除和增量拉取。
Iceberg 的特性主要是代碼抽象程度高，不綁定任何的 Engine。它暴露出來了非常核心的表層面的接口，可以非常方便的與 Spark/Flink 對接。然而 Delta 和 Hudi 基本上和 Spark 的耦合很重。如果想接入 Flink，相對比較難。

3. 我們選擇 Iceberg 的原因：

現(xiàn)在國內(nèi)的實時數(shù)倉建設(shè)圍繞 Flink 的情況會多一點。所以能夠基于 Flink 擴展生態(tài)，是我們選擇 Iceberg 一個比較重要的點。
國內(nèi)也有很多基于 Iceberg 開發(fā)的重要力量，比如騰訊團隊、Flink 官方團隊，他們的數(shù)據(jù)湖選型也是 Iceberg。目前他們在社區(qū)分別主導(dǎo) update 以及 Flink 的生態(tài)對接。

4. 接下來我們重點介紹一下 Iceberg：

 Iceberg 是一個開源的基于表格式的數(shù)據(jù)湖。關(guān)于 table format 再給大家詳細介紹下：

左側(cè)圖是一個抽象的數(shù)據(jù)處理系統(tǒng)，分別由 SQL 引擎、table format、文件集合以及分布式文件系統(tǒng)構(gòu)成。右側(cè)是對應(yīng)的現(xiàn)實中的組件，SQL 引擎比如 HiveServer、Impala、Spark 等等，table format 比如 Metastore 或者 Iceberg，文件集合主要有 Parquet 文件等，而分布式文件系統(tǒng)就是 HDFS。

對于 table format，我認為主要包含4個層面的含義，分別是表 schema 定義（是否支持復(fù)雜數(shù)據(jù)類型），表中文件的組織形式，表相關(guān)統(tǒng)計信息、表索引信息以及表的讀寫 API 實現(xiàn)。詳述如下：

• 表 schema 定義了一個表支持字段類型，比如 int、string、long 以及復(fù)雜數(shù)據(jù)類型等。

• 表中文件組織形式最典型的是 Partition 模式，是 Range Partition 還是 Hash Partition。

• Metadata 數(shù)據(jù)統(tǒng)計信息。

• 封裝了表的讀寫 API。上層引擎通過對應(yīng)的API讀取或者寫入表中的數(shù)據(jù)。

和 Iceberg 差不多相當?shù)囊粋€組件是 Metastore。不過 Metastore 是一個服務(wù)，而 Iceberg 就是一個 jar 包。這里就 Metastore 和 Iceberg 在表格式的4個方面分別進行一下對比介紹：

① 在 schema 層面上沒有任何區(qū)別：

都支持 int、string、bigint 等類型。

② partition 實現(xiàn)完全不同：

兩者在 partition 上有很大的不同：

metastore 中 partition 字段不能是表字段，因為 partition 字段本質(zhì)上是一個目錄結(jié)構(gòu)，不是用戶表中的一列數(shù)據(jù)?；?metastore，用戶想定位到一個 partition 下的所有數(shù)據(jù)，首先需要在 metastore 中定位出該 partition 對應(yīng)的所在目錄位置信息，然后再到 HDFS 上執(zhí)行l(wèi)ist命令獲取到這個分區(qū)下的所有文件，對這些文件進行掃描得到這個 partition 下的所有數(shù)據(jù)。

Iceberg 中 partition 字段就是表中的一個字段。Iceberg 中每一張表都有一個對應(yīng)的文件元數(shù)據(jù)表，文件元數(shù)據(jù)表中每條記錄表示一個文件的相關(guān)信息，這些信息中有一個字段是 partition 字段，表示這個文件所在的 partition。

很明顯，Iceberg 表根據(jù) partition 定位文件相比 metastore 少了一個步驟，就是根據(jù)目錄信息去 HDFS 上執(zhí)行 list 命令獲取分區(qū)下的文件。

試想，對于一個二級分區(qū)的大表來說，一級分區(qū)是小時時間分區(qū)，二級分區(qū)是一個枚舉字段分區(qū)，假如每個一級分區(qū)下有30個二級分區(qū)，那么這個表每天就會有24 * 30 = 720個分區(qū)?；?Metastore 的 partition 方案，如果一個 SQL 想基于這個表掃描昨天一天的數(shù)據(jù)的話，就需要向 Namenode 下發(fā)720次 list 請求，如果掃描一周數(shù)據(jù)或者一個月數(shù)據(jù)，請求數(shù)就更是相當夸張。這樣，一方面會導(dǎo)致 Namenode 壓力很大，一方面也會導(dǎo)致 SQL 請求響應(yīng)延遲很大。而基于 Iceberg 的 partition 方案，就完全沒有這個問題。

③ 表統(tǒng)計信息實現(xiàn)粒度不同：

Metastore 中一張表的統(tǒng)計信息是表/分區(qū)級別粒度的統(tǒng)計信息，比如記錄一張表中某一列的記錄數(shù)量、平均長度、為 null 的記錄數(shù)量、最大值最小值等。

Iceberg 中統(tǒng)計信息精確到文件粒度，即每個數(shù)據(jù)文件都會記錄所有列的記錄數(shù)量、平均長度、最大值最小值等。

很明顯，文件粒度的統(tǒng)計信息對于查詢中謂詞（即 where 條件）的過濾會更有效果。

④ 讀寫 API 實現(xiàn)不同：

metastore 模式下上層引擎寫好一批文件，調(diào)用 metastore 的 add partition 接口將這些文件添加到某個分區(qū)下。

Iceberg 模式下上層業(yè)務(wù)寫好一批文件，調(diào)用 iceberg 的 commit 接口提交本次寫入形成一個新的 snapshot 快照。這種提交方式保證了表的 ACID 語義。同時基于 snapshot 快照提交可以實現(xiàn)增量拉取實現(xiàn)。

總結(jié)下 Iceberg 相對于 Metastore 的優(yōu)勢：

• 新 partition 模式：避免了查詢時n次調(diào)用 namenode 的 list 方法，降低 namenode 壓力，提升查詢性能

• 新 metadata 模式：文件級別列統(tǒng)計信息可以用來根據(jù) where 字段進行文件過濾，很多場景下可以大大減少掃描文件數(shù)，提升查詢性能

• 新 API 模式：存儲批流一體

1. 流式寫入-增量拉?。ɑ?Iceberg 統(tǒng)一存儲模式可以同時滿足業(yè)務(wù)批量讀取以及增量訂閱需求）

1. 支持批流同時讀寫同一張表，統(tǒng)一表schema，任務(wù)執(zhí)行過程中不會出現(xiàn) FileNotFoundException

Iceberg 的提升體現(xiàn)在：

03 數(shù)據(jù)湖 Iceberg 社區(qū)現(xiàn)狀

目前 Iceberg 主要支持的計算引擎包括 Spark 2.4.5、Spark 3.x、Flink 1.11 以及 Presto。同時，一些運維工作比如 snapshot 過期、小文件合并、增量訂閱消費等功能都可以實現(xiàn)。

對于 Apache Flink 來說，Apache Iceberg 是 delta、iceberg、hudi 三個開源項目中最先完成 Flink 接入的開源項目。通過 Flink 來完成實時導(dǎo)入數(shù)據(jù)到 Iceberg 數(shù)據(jù)湖、通過 Flink batch 作業(yè)來讀取 Iceberg 數(shù)據(jù)，這兩個核心功能將在 Apache Iceberg 0.10.0 版本發(fā)布（預(yù)計將在10月底發(fā)布）。對 Flink+iceberg 集成工作感興趣的同學(xué)，可以參考 Apache Iceberg 社區(qū)的使用文檔。

https://github.com/apache/iceberg/blob/master/site/docs/flink.md

按照目前的研發(fā)進度，我們預(yù)計實時寫入和讀取 CDC 數(shù)據(jù)這個功能，將在 Iceberg 的0.11.0版本發(fā)布。

04 網(wǎng)易數(shù)據(jù)湖 Iceberg 實踐之路

Iceberg 針對目前的大數(shù)量的情況下，可以大大提升 ETL 任務(wù)執(zhí)行的效率，這主要得益于新 Partition 模式下不再需要請求 NameNode 分區(qū)信息，同時得益于文件級別統(tǒng)計信息模式下可以過濾很多不滿足條件的數(shù)據(jù)文件。

當前 Iceberg 社區(qū)僅支持 Spark 2.4.5，我們在這個基礎(chǔ)上做了更多計算引擎的適配工作。主要包括如下：

• 集成 Hive。可以通過 Hive 創(chuàng)建和刪除 iceberg 表，通過 HiveSQL 查詢 Iceberg 表中的數(shù)據(jù)。

• 集成 Impala。用戶可以通過 Impala 新建 iceberg 內(nèi)表外表，并通過 Impala 查詢 Iceberg 表中的數(shù)據(jù)。目前該功能已經(jīng)貢獻給 Impala 社區(qū)。

• 集成 Flink。已經(jīng)實現(xiàn)了 Flink 到 Iceberg 的 sink 實現(xiàn)，業(yè)務(wù)可以消費 kafka 中的數(shù)據(jù)將結(jié)果寫入到 Iceberg 中。同時我們基于 Flink 引擎實現(xiàn)了小文件異步合并的功能，這樣可以實現(xiàn) Flink 一邊寫數(shù)據(jù)文件，一邊執(zhí)行小文件的合并。基于 Iceberg 的小文件合并通過 commit 的方式提交，不需要刪除合并前的小文件，也就不會引起讀取任務(wù)的任何異常。

上述就是小編為大家分享的Flink+Iceberg 數(shù)據(jù)湖探索與實踐是怎樣的了，如果剛好有類似的疑惑，不妨參照上述分析進行理解。如果想知道更多相關(guān)知識，歡迎關(guān)注創(chuàng)新互聯(lián)行業(yè)資訊頻道。

文章名稱：Flink+Iceberg數(shù)據(jù)湖探索與實踐是怎樣的
文章出自：http://muchs.cn/article0/jpeooo.html

成都網(wǎng)站建設(shè)公司_創(chuàng)新互聯(lián)，為您提供品牌網(wǎng)站制作、手機網(wǎng)站建設(shè)、App開發(fā)、移動網(wǎng)站建設(shè)、網(wǎng)站建設(shè)、域名注冊

廣告

聲明：本網(wǎng)站發(fā)布的內(nèi)容（圖片、視頻和文字）以用戶投稿、用戶轉(zhuǎn)載內(nèi)容為主，如果涉及侵權(quán)請盡快告知，我們將會在第一時間刪除。文章觀點不代表本網(wǎng)站立場，如需處理請聯(lián)系客服。電話：028-86922220；郵箱：631063699@qq.com。內(nèi)容未經(jīng)允許不得轉(zhuǎn)載，或轉(zhuǎn)載時需注明來源： 創(chuàng)新互聯(lián)