Hudi 實(shí)踐 | 如何將數(shù)據(jù)更快導(dǎo)入 Apache Hudi？

1. 摘要

Apache Hudi除了支持insert和upsert外，還支持bulk_insert操作將數(shù)據(jù)攝入Hudi表，對(duì)于bulk_insert操作有不同的使用模式，本篇博客將闡述bulk_insert不同的模式以及與其他操作的比較。

Apache Hudi支持bulk_insert操作來(lái)將數(shù)據(jù)初始化至Hudi表中，該操作相比insert和upsert操作速度更快，效率更高。bulk_insert不會(huì)查看已存在數(shù)據(jù)的開(kāi)銷并且不會(huì)進(jìn)行小文件優(yōu)化。

bulk_insert按照以下原則提供了3種模式來(lái)滿足不同的需求

?如果數(shù)據(jù)布局良好，排序?qū)槲覀兲峁┝己玫膲嚎s和upsert性能。特別是記錄鍵具有某種排序（時(shí)間戳等）特征，則排序?qū)⒂兄谠趗psert期間裁剪大量文件，如果數(shù)據(jù)是按頻繁查詢的列排序的，那么查詢將利用parquet謂詞下推來(lái)裁剪數(shù)據(jù)，以確保更低的查詢延遲。?寫parquet文件是內(nèi)存密集型操作。當(dāng)將大量數(shù)據(jù)寫入一個(gè)也被劃分為1000個(gè)分區(qū)的表中時(shí)，如果不進(jìn)行任何排序，寫入程序可能必須保持1000個(gè)parquet寫入器處于打開(kāi)狀態(tài)，同時(shí)會(huì)產(chǎn)生不可持續(xù)的內(nèi)存壓力，并最終導(dǎo)致崩潰。

?在批量導(dǎo)入數(shù)據(jù)時(shí)，最好控制好少的文件個(gè)數(shù)，以避免以后寫入和查詢時(shí)的元數(shù)據(jù)開(kāi)銷。

3種開(kāi)箱即用的模式為：PARTITION_SORT、GLOBAL_SORT、NONE

2. 配置

可以通過(guò)hoodie.bulkinsert.sort.mode^[1]配置項(xiàng)來(lái)設(shè)置上述模式（NONE, GLOBAL_SORT , PARTITION_SORT），默認(rèn)值為GLOBAL_SORT。

3. 不同模式

3.1 GLOBAL_SORT（全局排序）

顧名思義，Hudi在輸入分區(qū)中對(duì)記錄進(jìn)行全局排序，從而在索引查找過(guò)程中最大化使用鍵范圍修剪的文件數(shù)量，以便提升upsert性能。這是因?yàn)槊總€(gè)文件都具有非重疊的鍵的最小值和最大值，這在鍵具有某些排序特征（例如基于時(shí)間的前綴）時(shí)非常有用。假設(shè)我們?cè)谌魏谓o定的時(shí)間都在單個(gè)輸出分區(qū)路徑上寫入單個(gè)parquet文件，此模式在大分區(qū)寫入期間有助于控制內(nèi)存壓力。同樣由于全局排序，每個(gè)小表分區(qū)路徑將從最多有兩個(gè)分區(qū)寫入，因此只包含2個(gè)文件。該模式是Hudi中進(jìn)行bulk_insert操作的默認(rèn)模式。

3.2 PARTITION_SORT（分區(qū)排序）

在這種排序模式下將對(duì)給定spark分區(qū)內(nèi)的記錄進(jìn)行排序，但是給定的spark分區(qū)可能包含來(lái)自不同表分區(qū)的記錄，因此即使我們?cè)诿總€(gè)spark分區(qū)內(nèi)進(jìn)行排序，也可能會(huì)在產(chǎn)生大量文件，因?yàn)榻o定表分區(qū)的記錄可能會(huì)分布在許多spark分區(qū)中。在寫入器實(shí)際寫入時(shí)可能不會(huì)同時(shí)打開(kāi)太多文件，因?yàn)槲覀冊(cè)谝苿?dòng)到下一個(gè)文件之前關(guān)閉了該文件（記錄在spark分區(qū)中排序），因此可能沒(méi)有太大的內(nèi)存壓力。

3.3 NONE

在此模式下，不會(huì)對(duì)用戶記錄進(jìn)行任何轉(zhuǎn)換（如排序），將數(shù)據(jù)原樣委托給寫入器。因此在將大量數(shù)據(jù)寫入分區(qū)為1000個(gè)分區(qū)的表中時(shí)，寫入程序可能必須保持1000個(gè)parquet寫入程序處于打開(kāi)狀態(tài)，同時(shí)可能會(huì)產(chǎn)生較大內(nèi)存壓力，有可能導(dǎo)致崩潰，因此該模式下會(huì)有較大的內(nèi)存開(kāi)銷。此外給定文件的最小-最大范圍可能非常寬（未排序的記錄），因此后續(xù)的upsert會(huì)在索引查找期間從大量文件中讀取bloom filter（布隆過(guò)濾器）。由于記錄沒(méi)有排序，并且每個(gè)寫入器可以跨N個(gè)表分區(qū)獲取記錄，因此這種模式可能會(huì)導(dǎo)致在bulk_insert結(jié)束時(shí)產(chǎn)生大量文件。由于有大量的小文件，這也可能會(huì)影響upsert或查詢性能。

4. 用戶自定義Partitioner

如果上述模式都不能滿足需求，用戶可以自定義實(shí)現(xiàn)partitioner^[2]來(lái)滿足業(yè)務(wù)需求。

5. 性能測(cè)試

不同模式下簡(jiǎn)單benchmark性能差異如下

說(shuō)明：該基準(zhǔn)測(cè)試使用不同的排序模式將1000萬(wàn)條記錄批量插入hudi，然后upsert100W個(gè)條記錄（原始數(shù)據(jù)集大小的10%）。

顯而易見(jiàn)，NONE模式對(duì)批量導(dǎo)入性能最佳，因?yàn)樗簧婕叭魏闻判?。與NONE模式相比，GLOBAL_SORT相比NONE模式開(kāi)銷約為15%。PARTITION_SORT相比NONE模式開(kāi)銷約為4%，因?yàn)橐采婕暗綄?duì)記錄的排序操作。但是要注意的是后面的upsert性能。如前所述，與其他兩種排序模式相比全局排序具有許多優(yōu)勢(shì)，GLOBAL_SORT相比NONE upsert性能高40%。PARTITION_SORT相比NONE模式有約5%的改進(jìn)，這是由于大量小文件開(kāi)銷導(dǎo)致。

6. 總結(jié)

希望這個(gè)博客能讓你很好地了解bulk_insert中的不同模式以及何時(shí)使用哪種模式。如果想?yún)⑴cHudi社區(qū)，請(qǐng)點(diǎn)擊這里^[3]。

推薦閱讀

Flink + Hudi，構(gòu)架倉(cāng)湖一體化解決方案

基于Apache Hudi 的CDC數(shù)據(jù)入湖

內(nèi)附PPT下載｜萬(wàn)字干貨！阿里云基于Apache Hudi構(gòu)建Lakehouse實(shí)踐探索

一文徹底掌握Apache Hudi異步Clustering部署

使用 Flink Hudi 構(gòu)建流式數(shù)據(jù)湖

引用鏈接

[1] hoodie.bulkinsert.sort.mode: https://hudi.apache.org/docs/configurations.html#withBulkInsertSortMode
[2] partitioner: https://github.com/apache/hudi/blob/master/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/BulkInsertPartitioner.java
[3] 這里: https://hudi.apache.org/contribute/get-involved