背景及痛點(diǎn)
業(yè)務(wù)背景
同程藝龍是一個(gè)提供機(jī)票，住宿，交通等服務(wù)的在線旅游服務(wù)平臺(tái)，目前我所在的部門(mén)屬于公司的研發(fā)部門(mén)，主要職責(zé)是為公司內(nèi)其他業(yè)務(wù)部門(mén)提供一些基礎(chǔ)服務(wù)，我們的大數(shù)據(jù)系統(tǒng)主要承接的業(yè)務(wù)是部門(mén)內(nèi)的一些大數(shù)據(jù)相關(guān)的數(shù)據(jù)統(tǒng)計(jì)、分析工作等，數(shù)據(jù)來(lái)源有網(wǎng)關(guān)日志數(shù)據(jù)，服務(wù)器監(jiān)控?cái)?shù)據(jù)，k8s容器的相關(guān)日志數(shù)據(jù)，app的打點(diǎn)日志, mysql的binlog日志等。我們的主要的大數(shù)據(jù)的任務(wù)就是基于這些日志構(gòu)建實(shí)時(shí)報(bào)表，提供基于presto的報(bào)表展示和即時(shí)查詢服務(wù)，以及基于flink開(kāi)發(fā)一些實(shí)時(shí)、批處理任務(wù)，為業(yè)務(wù)方提供準(zhǔn)確及時(shí)的數(shù)據(jù)支撐。
原架構(gòu)方案
由于我們所有的原始數(shù)據(jù)都是存儲(chǔ)在kafka的，所以原來(lái)的技術(shù)架構(gòu)就是首先是flink任務(wù)消費(fèi)kafka的數(shù)據(jù)，經(jīng)過(guò)flink sql或者flink jar的各種處理之后實(shí)時(shí)寫(xiě)入hive，其中絕大部分任務(wù)都是flink sql任務(wù)，因?yàn)槲艺J(rèn)為sql開(kāi)發(fā)相對(duì)代碼要簡(jiǎn)單的多，并且維護(hù)方便、好理解,所以能用sql寫(xiě)的都盡量用sql來(lái)寫(xiě)。
提交flink的平臺(tái)使用的是zeppelin，其中提交flink sql任務(wù)是zeppelin自帶的功能，提交jar包任務(wù)是我自己基于application模式開(kāi)發(fā)的zeppelin插件。
對(duì)于落地到hive的數(shù)據(jù)，使用開(kāi)源的報(bào)表系統(tǒng)metabase (底層使用presto) 提供實(shí)時(shí)報(bào)表展示、定時(shí)發(fā)送郵件報(bào)表，以及自定義sql查詢服務(wù)。由于業(yè)務(wù)對(duì)數(shù)據(jù)的實(shí)時(shí)性要求比較高，希望數(shù)據(jù)能盡快的展示出來(lái)，所以我們很多的flink流式任務(wù)的checkpoint設(shè)置為1分鐘，數(shù)據(jù)格式采用的是orc格式。
痛點(diǎn)
由于采用的是列式存儲(chǔ)格式orc，無(wú)法像行式存儲(chǔ)格式那樣進(jìn)行追加操作，所以不可避免的產(chǎn)生了一個(gè)大數(shù)據(jù)領(lǐng)域非常常見(jiàn)且非常棘手的問(wèn)題，即hdfs小文件問(wèn)題。
開(kāi)始的時(shí)候我們的小文件解決方案是自己寫(xiě)的一個(gè)小文件壓縮工具，定期的去合并，我們的hive分區(qū)一般都是天級(jí)別的，所以這個(gè)工具的原理就是每天的凌晨啟動(dòng)一個(gè)定時(shí)任務(wù)去壓縮昨天的數(shù)據(jù)，首先把昨天的數(shù)據(jù)寫(xiě)入一個(gè)臨時(shí)文件夾，壓縮完，和原來(lái)的數(shù)據(jù)進(jìn)行記錄數(shù)的比對(duì)檢驗(yàn)，數(shù)據(jù)條數(shù)一致之后，用壓縮后的數(shù)據(jù)覆蓋原來(lái)的數(shù)據(jù)，但是由于無(wú)法保證事務(wù)，所以出現(xiàn)了很多的問(wèn)題：
壓縮的同時(shí)由于延遲數(shù)據(jù)的到來(lái)導(dǎo)致昨天的hive分區(qū)又有數(shù)據(jù)寫(xiě)入了，檢驗(yàn)就會(huì)失敗，導(dǎo)致合并小文件失敗。
替換舊數(shù)據(jù)的操作是沒(méi)有事務(wù)保證的，如果替換的過(guò)程中舊分區(qū)有新的數(shù)據(jù)寫(xiě)入，就會(huì)覆蓋新寫(xiě)入的數(shù)據(jù)，造成數(shù)據(jù)丟失。
沒(méi)有事務(wù)的支持，無(wú)法實(shí)時(shí)的合并當(dāng)前分區(qū)的數(shù)據(jù)，只能合并壓縮前一個(gè)分區(qū)的，最新的分區(qū)數(shù)據(jù)仍然有小文件的問(wèn)題，導(dǎo)致最新數(shù)據(jù)查詢性能提高不了。
flink+iceberg的落地
iceberg技術(shù)調(diào)研
所以基于以上的hdfs小文件、查詢慢等問(wèn)題，結(jié)合我們的現(xiàn)狀，我調(diào)研了目前市面上的數(shù)據(jù)湖技術(shù)：delta、Apache Iceberg和Apache Hudi，考慮了目前數(shù)據(jù)湖框架支持的功能和以后的社區(qū)規(guī)劃，最終我們是選擇了iceberg，其中考慮的原因有以下幾方面：
-
iceberg深度集成flink
前面講到，我們的絕大部分任務(wù)都是flink任務(wù)，包括批處理任務(wù)和流處理任務(wù)，目前這三個(gè)數(shù)據(jù)湖框架，iceberg是集成flink做的最完善的，如果采用iceberg替代hive之后，遷移的成本非常小，對(duì)用戶幾乎是無(wú)感知的，
比如我們?cè)瓉?lái)的sql是這樣的，
INSERT INTO hive_catalog.db.hive_table SELECT * FROM kafka_table遷移到iceberg以后，只需要修改catalog就行了.
INSERT INTO iceberg_catalog.db.iceberg_table SELECT * FROM kafka_tablepresto查詢也是和這個(gè)類似，只需要修改catalog就行了。
iceberg的設(shè)計(jì)架構(gòu)使得查詢更快
[圖片上傳失敗...(image-8119df-1616550735167)]
在iceberg的設(shè)計(jì)架構(gòu)中，manifest文件存儲(chǔ)了分區(qū)相關(guān)信息、data files的相關(guān)統(tǒng)計(jì)信息（max/min）等，去查詢一些大的分區(qū)的數(shù)據(jù)，就可以直接定位到所要的數(shù)據(jù)，而不是像hive一樣去list整個(gè)hdfs文件夾，時(shí)間復(fù)雜度從O(n)降到了O(1)，使得一些大的查詢速度有了明顯的提升，在Iceberg PMC Chair Ryan Blue的演講中，我們看到命中filter的任務(wù)執(zhí)行時(shí)間從61.5小時(shí)降到了22分鐘。
- 使用flink sql將cdc數(shù)據(jù)寫(xiě)入iceberg
flink cdc提供了直接讀取MySQL binlog的方式，相對(duì)以前需要使用canal讀取binlog寫(xiě)入kafka，然后再去消費(fèi)kafka數(shù)據(jù)。少了兩個(gè)組件的維護(hù)，鏈路減少了，節(jié)省了維護(hù)的成本和出錯(cuò)的概率。并且可以實(shí)現(xiàn)導(dǎo)入全量數(shù)據(jù)和增量數(shù)據(jù)的完美對(duì)接，所以使用flink sql將MySQL binlog數(shù)據(jù)導(dǎo)入iceberg來(lái)做mysql->iceberg的導(dǎo)入將會(huì)是一件非常有意義的事情。
此外對(duì)于我們最初的壓縮小文件的需求，雖然iceberg目前還無(wú)法實(shí)現(xiàn)自動(dòng)壓縮，但是它提供了一個(gè)批處理任務(wù)，已經(jīng)能滿足我們的需求。
hive表遷移iceberg表
- 遷移準(zhǔn)備工作
目前我們的所有數(shù)據(jù)都是存儲(chǔ)在hive表的，在驗(yàn)證完iceberg之后，我們決定將hive的數(shù)據(jù)遷移到iceberg，所以我寫(xiě)了一個(gè)工具，可以使用hive的數(shù)據(jù)，然后新建一個(gè)iceberg表，為其建立相應(yīng)的元數(shù)據(jù)，但是測(cè)試的時(shí)候發(fā)現(xiàn)，如果采用這種方式，需要把寫(xiě)入hive的程序停止，因?yàn)槿绻鹖ceberg和hive使用同一個(gè)數(shù)據(jù)文件，而壓縮程序會(huì)不斷地壓縮iceberg表的小文件，壓縮完之后，不會(huì)馬上刪除舊數(shù)據(jù)，所以hive表就會(huì)查到雙份的數(shù)據(jù)，故我們采用雙寫(xiě)的策略，原來(lái)寫(xiě)入hive的程序不動(dòng)，新啟動(dòng)一套程序?qū)懭雐ceberg，這樣能對(duì)iceberg表觀察一段時(shí)間。還能和原來(lái)hive中的數(shù)據(jù)進(jìn)行比對(duì)，來(lái)驗(yàn)證程序的正確性。
經(jīng)過(guò)一段時(shí)間觀察，每天將近幾十億條數(shù)據(jù)、壓縮后幾個(gè)T大小的hive表和iceberg表，一條數(shù)據(jù)也不差。所以在最終對(duì)比數(shù)據(jù)沒(méi)有問(wèn)題之后，把hive表停止寫(xiě)入，使用新的iceberg表。
- 遷移工具
我將這個(gè)hive表遷移iceberg表的工具做成了一個(gè)基于flink batch job的iceberg Action，提交了社區(qū)，不過(guò)目前還沒(méi)合并：https://github.com/apache/iceberg/pull/2217 ， 這個(gè)功能的思路是使用hive原始的數(shù)據(jù)不動(dòng)，然后新建一個(gè)iceberg table，然后為這個(gè)新的iceberg table 生成對(duì)應(yīng)的元數(shù)據(jù)，大家有需要的話可以先看看。
此外，iceberg社區(qū)，還有一個(gè)把現(xiàn)有的數(shù)據(jù)遷移到已存在的iceberg table的工具，類似hive的LOAD DATA INPATH ... INTO TABLE,是用spark的存儲(chǔ)過(guò)程做的，大家也可以關(guān)注下：https://github.com/apache/iceberg/pull/2210
iceberg優(yōu)化實(shí)踐
壓縮小文件
目前壓縮小文件是采用的一個(gè)額外批任務(wù)來(lái)進(jìn)行的，Iceberg提供了一個(gè)spark版本的action，我在做功能測(cè)試的時(shí)候發(fā)現(xiàn)了一些問(wèn)題，此外我對(duì)spark也不是非常熟悉，擔(dān)心出了問(wèn)題不好排查，所以參照spark版本的自己實(shí)現(xiàn)了一個(gè)flink版本，并修復(fù)了一些bug，進(jìn)行了一些功能的優(yōu)化。
由于我們的iceberg的元數(shù)據(jù)都是存儲(chǔ)在hive中的，也就是我們使用了HiveCatalog,所以壓縮程序的邏輯是我把hive中所有的iceberg表全部都查出來(lái)，依次壓縮。壓縮沒(méi)有過(guò)濾條件，不管是分區(qū)表還是非分區(qū)表，都進(jìn)行全表的壓縮。這樣做是為了處理某些使用eventtime的flink任務(wù)，如果有延遲的數(shù)據(jù)的到來(lái)。就會(huì)把數(shù)據(jù)寫(xiě)入以前的分區(qū)，如果不是全表壓縮只壓縮當(dāng)天分區(qū)的話，新寫(xiě)入的其他天的數(shù)據(jù)就不會(huì)被壓縮。
之所以沒(méi)有開(kāi)啟定時(shí)任務(wù)來(lái)壓縮，是因?yàn)楸热缥叶〞r(shí)五分鐘壓縮一個(gè)表，如果五分鐘之內(nèi)這個(gè)壓縮任務(wù)沒(méi)完成，沒(méi)有提交新的snapshot，下一個(gè)定時(shí)任務(wù)又開(kāi)啟了，就會(huì)把上一個(gè)沒(méi)有完成的壓縮任務(wù)中的數(shù)據(jù)重新壓縮一次，所以每個(gè)表依次壓縮的策略可以保證某一時(shí)刻一個(gè)表只有一個(gè)任務(wù)在壓縮。
代碼示例參考：
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
Actions.forTable(env, table)
.rewriteDataFiles()
//.maxParallelism(parallelism)
//.filter(Expressions.equal("day", day))
//.targetSizeInBytes(targetSizeInBytes)
.execute();
目前系統(tǒng)運(yùn)行穩(wěn)定，已經(jīng)完成了幾萬(wàn)次任務(wù)的壓縮
[圖片上傳失敗...(image-12f54b-1616550735166)]
注意：
不過(guò)目前對(duì)于新發(fā)布的iceberg 0.11來(lái)說(shuō)，還有一個(gè)已知的bug，就是當(dāng)壓縮前的文件大小大于要壓縮的大?。╰argetSizeInBytes）的時(shí)候，會(huì)造成數(shù)據(jù)丟失，其實(shí)這個(gè)問(wèn)題我在最開(kāi)始測(cè)試小文件壓縮的時(shí)候就發(fā)現(xiàn)了，并且提了一個(gè)pr，我的策略是大于目標(biāo)文件的數(shù)據(jù)文件不參與壓縮，不過(guò)這個(gè)pr沒(méi)有合并到0.11版本中，后來(lái)社區(qū)另外一個(gè)兄弟也發(fā)現(xiàn)了相同的問(wèn)題，提交了一個(gè)pr（ https://github.com/apache/iceberg/pull/2196 ） ，策略是將這個(gè)大文件拆分到目標(biāo)文件大小，目前已經(jīng)合并到master，會(huì)在下一個(gè)bug fix版本0.11.1中發(fā)布。
查詢優(yōu)化
- 批處理定時(shí)任務(wù)
目前對(duì)于定時(shí)調(diào)度中的批處理任務(wù)，flink的sql客戶端還沒(méi)hive那樣做的很完善，比如執(zhí)行hive -f來(lái)執(zhí)行一個(gè)文件。而且不同的任務(wù)需要不同的資源，并行度等。 所以我自己封裝了一個(gè)flink程序，通過(guò)調(diào)用這個(gè)程序來(lái)進(jìn)行處理，讀取一個(gè)指定文件里面的sql，來(lái)提交批任務(wù)。在命令行控制任務(wù)的資源和并行度等。
/home/flink/bin/flink run -p 10 -m yarn-cluster /home/work/iceberg-scheduler.jar my.sql
- 優(yōu)化
批任務(wù)的查詢這塊，我做了一些優(yōu)化工作，比如limit下推，filter下推，查詢并行度推斷等，可以大大提高查詢的速度，這些優(yōu)化都已經(jīng)推回給社區(qū)，并且在iceberg 0.11版本中發(fā)布。
運(yùn)維管理
- 清理orphan文件
- 定時(shí)任務(wù)刪除
在使用iceberg的過(guò)程中，有時(shí)候會(huì)有這樣的情況，我提交了一個(gè)flink任務(wù)，由于各種原因，我把它給停了，這個(gè)時(shí)候iceberg還沒(méi)提交相應(yīng)的快照。還有由于一些異常導(dǎo)致程序失敗，就會(huì)產(chǎn)生一些不在iceberg元數(shù)據(jù)里面的孤立的數(shù)據(jù)文件，這些文件對(duì)iceberg來(lái)說(shuō)是不可達(dá)的，也是沒(méi)用的。所以我們需要像jvm的垃圾回收一樣來(lái)清理這些文件。
目前iceberg提供了一個(gè)spark版本的action來(lái)進(jìn)行處理這些沒(méi)用的文件，我們采取的策略和壓縮小文件一樣，獲取hive中的所有的iceberg表。每隔一個(gè)小時(shí)執(zhí)行一次定時(shí)任務(wù)來(lái)刪除這些沒(méi)用的文件。
SparkSession spark = ......
Actions.forTable(spark, table)
.removeOrphanFiles()
//.deleteWith(...)
.execute();
- 踩坑
在程序運(yùn)行過(guò)程中出現(xiàn)了正常的數(shù)據(jù)文件被刪除的問(wèn)題，經(jīng)過(guò)調(diào)研，由于我的快照保留設(shè)置是一小時(shí)，這個(gè)清理程序清理時(shí)間也是設(shè)置一個(gè)小時(shí)，通過(guò)日志發(fā)現(xiàn)是這個(gè)清理程序刪除了正常的數(shù)據(jù)。查了查代碼，應(yīng)該是他們?cè)O(shè)置了一樣的時(shí)間，在清理孤立文件的時(shí)候，有其他程序正在讀取這個(gè)要expired的snapshot，導(dǎo)致刪除了正常的數(shù)據(jù)。最后把這個(gè)清理程序的清理時(shí)間改成默認(rèn)的三天，沒(méi)有再出現(xiàn)刪除數(shù)據(jù)文件的問(wèn)題。 當(dāng)然，為了保險(xiǎn)起見(jiàn)，我們可以覆蓋原來(lái)的刪除文件的方法，改成將文件到一個(gè)備份文件夾，檢查沒(méi)有問(wèn)題之后，手工刪除。
- 快照過(guò)期處理
我們的快照過(guò)期策略，我是和壓縮小文件的批處理任務(wù)寫(xiě)在一起的，壓縮完小文件之后，進(jìn)行表的快照過(guò)期處理，目前保留的時(shí)間是一個(gè)小時(shí)，這是因?yàn)閷?duì)于有一些比較大的表，分區(qū)比較多，而且checkpoint比較短，如果保留的快照過(guò)長(zhǎng)的話，還是會(huì)保留過(guò)多小文件，我們暫時(shí)沒(méi)有查詢歷史快照的需求，所以我將快照的保留時(shí)間設(shè)置了一個(gè)小時(shí)。
long olderThanTimestamp = System.currentTimeMillis() - TimeUnit.HOURS.toMillis(1);
table.expireSnapshots()
// .retainLast(20)
.expireOlderThan(olderThanTimestamp)
.commit();
- 數(shù)據(jù)管理
寫(xiě)入了數(shù)據(jù)之后，有時(shí)候我想查看一下相應(yīng)的快照下面有多少數(shù)據(jù)文件，直接查詢hdfs你不知道哪個(gè)是有用的，哪個(gè)是沒(méi)用的。所以需要有對(duì)應(yīng)的管理工具。目前flink這塊還不太成熟，我們可以使用spark3提供的工具來(lái)查看。
- DDL
目前create table 這些操作我們是通過(guò)flink sql client來(lái)做的。 其他相關(guān)的ddl的操作可以使用spark來(lái)做：
https://iceberg.apache.org/spark/#ddl-commands
- DML
一些相關(guān)的數(shù)據(jù)的操作，比如刪除數(shù)據(jù)等可以通過(guò)spark來(lái)實(shí)現(xiàn)，presto目前只支持分區(qū)級(jí)別的刪除功能。
- show partitions & show create table
在我們操作hive的時(shí)候，有一些很常用的操作，比如show partitions、 show create table 等，這些目前flink還沒(méi)有支持，所以在操作iceberg的時(shí)候就很不方便，我們自己基于flink 1.12做了修改，不過(guò)目前還沒(méi)有完全提交到社區(qū)，后續(xù)有時(shí)間會(huì)提交到flink 和iceberg 社區(qū)。
后續(xù)工作
flink sql接入cdc數(shù)據(jù)到iceberg
目前在我們內(nèi)部的版本中，我已經(jīng)測(cè)試通過(guò)可以使用flink sql 將cdc數(shù)據(jù)（比如mysql binlog）寫(xiě)入iceberg，社區(qū)的版本中實(shí)現(xiàn)該功能還需要做一些工作，我也提交了一些相關(guān)的PR來(lái)推進(jìn)這個(gè)工作。
使用sql進(jìn)行刪除和更新
對(duì)于copy-on-write表，我們可以使用spark sql來(lái)進(jìn)行行級(jí)的刪除和刪除。具體的支持的語(yǔ)法可以參考源碼中的測(cè)試類：org.apache.iceberg.spark.extensions.TestDelete & org.apache.iceberg.spark.extensions.TestUpdate，這些功能我在測(cè)試環(huán)境測(cè)試是可以的，但是還沒(méi)有來(lái)得及更新到生產(chǎn)。
使用flink sql進(jìn)行streaming read
在工作中會(huì)有一些這樣的場(chǎng)景，由于數(shù)據(jù)比較大，kafka的數(shù)據(jù)只存了較短的時(shí)間，如果很不幸，我因?yàn)槌绦驅(qū)戝e(cuò)了或者業(yè)務(wù)變更等原因，想從更早的時(shí)間來(lái)消費(fèi)，就無(wú)能為力了。
當(dāng)引入了iceberg的streaming read之后，這些問(wèn)題就可以解決了，因?yàn)閕ceberg存儲(chǔ)了所有的數(shù)據(jù)，當(dāng)然這里有一個(gè)前提就是對(duì)于數(shù)據(jù)沒(méi)有要求特別精確，比如達(dá)到秒級(jí)別，因?yàn)槟壳癴link寫(xiě)入iceberg的事務(wù)提交是基于flink checkpoint間隔的。
收益及總結(jié)
經(jīng)過(guò)對(duì)iceberg大概一個(gè)季度的調(diào)研，測(cè)試，優(yōu)化和bug修復(fù)，我們將現(xiàn)有的hive表都遷移到了iceberg，完美解決了原來(lái)的所有的痛點(diǎn)問(wèn)題，目前系統(tǒng)穩(wěn)定運(yùn)行，而且相對(duì)hive得到了很多的收益：
- flink寫(xiě)入的資源減少
舉一個(gè)例子，默認(rèn)配置下，原來(lái)一個(gè)flink讀取kafka寫(xiě)入hive的任務(wù)，需要60個(gè)并行度才不會(huì)讓kafka產(chǎn)生積壓。改成寫(xiě)入iceberg之后，只需要20個(gè)并行度就夠了. - 查詢速度變快
前面我們講到iceberg查詢的時(shí)候不會(huì)像hive一樣去list整個(gè)文件夾來(lái)獲取分區(qū)數(shù)據(jù)，而是先從manifest文件中獲取相關(guān)數(shù)據(jù)，查詢的性能得到了顯著的提升，一些大的報(bào)表的查詢速度從50秒提高到30秒。 - 并發(fā)讀寫(xiě)
由于iceberg的事務(wù)支持，我們可以實(shí)現(xiàn)對(duì)一個(gè)表進(jìn)行并發(fā)讀寫(xiě)，flink流式數(shù)據(jù)實(shí)時(shí)入湖，壓縮程序同時(shí)壓縮小文件，清理過(guò)期文件和快照的程序同時(shí)清理無(wú)用的文件，這樣就能更及時(shí)的提供數(shù)據(jù)，做到分鐘級(jí)的延遲，查詢最新分區(qū)數(shù)據(jù)的速度大大加快了，并且由于iceberg的ACID特性可以保證數(shù)據(jù)的準(zhǔn)確性。 - time travel
可以回溯查詢以前某一時(shí)刻的數(shù)據(jù)。
總結(jié)一下，我們目前可以實(shí)現(xiàn)使用flink sql 對(duì)iceberg進(jìn)行批、流的讀寫(xiě)，并可以對(duì)小文件進(jìn)行實(shí)時(shí)的壓縮,使用spark sql做一些delete和update工作以及一些DDL操作，后續(xù)可以使用flink sql 將cdc的數(shù)據(jù)寫(xiě)入iceberg，目前對(duì)iceberg的所有的優(yōu)化和bug fix，我已經(jīng)貢獻(xiàn)給社區(qū)。由于筆者水平有限，有時(shí)候也難免有錯(cuò)誤，還請(qǐng)大家不吝賜教。
