1. 背景

Apache Hudi將流處理帶到大數(shù)據(jù)，相比傳統(tǒng)批處理效率高一個(gè)數(shù)量級(jí)，提供了更新鮮的數(shù)據(jù)。在數(shù)據(jù)湖/倉(cāng)庫(kù)中，需要在攝取速度和查詢(xún)性能之間進(jìn)行權(quán)衡，數(shù)據(jù)攝取通常更喜歡小文件以改善并行性并使數(shù)據(jù)盡快可用于查詢(xún)，但很多小文件會(huì)導(dǎo)致查詢(xún)性能下降。在攝取過(guò)程中通常會(huì)根據(jù)時(shí)間在同一位置放置數(shù)據(jù)，但如果把查詢(xún)頻繁的數(shù)據(jù)放在一起時(shí)，查詢(xún)引擎的性能會(huì)更好，大多數(shù)系統(tǒng)都傾向于支持獨(dú)立的優(yōu)化來(lái)提高性能，以解決未優(yōu)化的數(shù)據(jù)布局的限制。本博客介紹了一種稱(chēng)為Clustering[RFC-19]的服務(wù)，該服務(wù)可重新組織數(shù)據(jù)以提高查詢(xún)性能，也不會(huì)影響攝取速度。

2. Clustering架構(gòu)

Hudi通過(guò)其寫(xiě)入客戶(hù)端API提供了不同的操作，如insert/upsert/bulk_insert來(lái)將數(shù)據(jù)寫(xiě)入Hudi表。為了能夠在文件大小和攝取速度之間進(jìn)行權(quán)衡，Hudi提供了一個(gè)hoodie.parquet.small.file.limit配置來(lái)設(shè)置最小文件大小。用戶(hù)可以將該配置設(shè)置為0以強(qiáng)制新數(shù)據(jù)寫(xiě)入新的文件組，或設(shè)置為更高的值以確保新數(shù)據(jù)被"填充"到現(xiàn)有小的文件組中，直到達(dá)到指定大小為止，但其會(huì)增加攝取延遲。

為能夠支持快速攝取的同時(shí)不影響查詢(xún)性能，我們引入了Clustering服務(wù)來(lái)重寫(xiě)數(shù)據(jù)以?xún)?yōu)化Hudi數(shù)據(jù)湖文件的布局。

Clustering服務(wù)可以異步或同步運(yùn)行，Clustering會(huì)添加了一種新的REPLACE操作類(lèi)型，該操作類(lèi)型將在Hudi元數(shù)據(jù)時(shí)間軸中標(biāo)記Clustering操作。

總體而言Clustering分為兩個(gè)部分：

•調(diào)度Clustering：使用可插拔的Clustering策略創(chuàng)建Clustering計(jì)劃。•執(zhí)行Clustering：使用執(zhí)行策略處理計(jì)劃以創(chuàng)建新文件并替換舊文件。

2.1 調(diào)度Clustering

調(diào)度Clustering會(huì)有如下步驟

•識(shí)別符合Clustering條件的文件：根據(jù)所選的Clustering策略，調(diào)度邏輯將識(shí)別符合Clustering條件的文件。•根據(jù)特定條件對(duì)符合Clustering條件的文件進(jìn)行分組。每個(gè)組的數(shù)據(jù)大小應(yīng)為targetFileSize的倍數(shù)。分組是計(jì)劃中定義的"策略"的一部分。此外還有一個(gè)選項(xiàng)可以限制組大小，以改善并行性并避免混排大量數(shù)據(jù)。•最后將Clustering計(jì)劃以avro元數(shù)據(jù)格式保存到時(shí)間線。

2.2 運(yùn)行Clustering

•讀取Clustering計(jì)劃，并獲得clusteringGroups，其標(biāo)記了需要進(jìn)行Clustering的文件組。•對(duì)于每個(gè)組使用strategyParams實(shí)例化適當(dāng)?shù)牟呗灶?lèi)（例如：sortColumns），然后應(yīng)用該策略重寫(xiě)數(shù)據(jù)。•創(chuàng)建一個(gè)REPLACE提交，并更新HoodieReplaceCommitMetadata中的元數(shù)據(jù)。

Clustering服務(wù)基于Hudi的MVCC設(shè)計(jì)，允許繼續(xù)插入新數(shù)據(jù)，而Clustering操作在后臺(tái)運(yùn)行以重新格式化數(shù)據(jù)布局，從而確保并發(fā)讀寫(xiě)者之間的快照隔離。

注意：現(xiàn)在對(duì)表進(jìn)行Clustering時(shí)還不支持更新，將來(lái)會(huì)支持并發(fā)更新。

2.3 Clustering配置

使用Spark可以輕松設(shè)置內(nèi)聯(lián)Clustering，參考如下示例

import org.apache.hudi.QuickstartUtils._</code><code>import scala.collection.JavaConversions._</code><code>import org.apache.spark.sql.SaveMode._</code><code>import org.apache.hudi.DataSourceReadOptions._</code><code>import org.apache.hudi.DataSourceWriteOptions._</code><code>import org.apache.hudi.config.HoodieWriteConfig._</code><code>val df =  //generate data frame</code><code>df.write.format("org.apache.hudi").</code><code>        options(getQuickstartWriteConfigs).</code><code>        option(PRECOMBINE_FIELD_OPT_KEY, "ts").</code><code>        option(RECORDKEY_FIELD_OPT_KEY, "uuid").</code><code>        option(PARTITIONPATH_FIELD_OPT_KEY, "partitionpath").</code><code>        option(TABLE_NAME, "tableName").</code><code>        option("hoodie.parquet.small.file.limit", "0").</code><code>        option("hoodie.clustering.inline", "true").</code><code>        option("hoodie.clustering.inline.max.commits", "4").</code><code>        option("hoodie.clustering.plan.strategy.target.file.max.bytes", "1073741824").</code><code>        option("hoodie.clustering.plan.strategy.small.file.limit", "629145600").</code><code>        option("hoodie.clustering.plan.strategy.sort.columns", "column1,column2"). //optional, if sorting is needed as part of rewriting data</code><code>        mode(Append).</code><code>        save("dfs://location");

對(duì)于設(shè)置更高級(jí)的異步Clustering管道，參考此處示例。

3. 表查詢(xún)性能

我們使用生產(chǎn)環(huán)境表的一個(gè)分區(qū)創(chuàng)建了一個(gè)數(shù)據(jù)集，該表具有約2000萬(wàn)條記錄，約200GB，數(shù)據(jù)集具有多個(gè)session_id的行。用戶(hù)始終使用會(huì)話謂詞查詢(xún)數(shù)據(jù)，單個(gè)會(huì)話的數(shù)據(jù)會(huì)分布在多個(gè)數(shù)據(jù)文件中，因?yàn)閿?shù)據(jù)攝取會(huì)根據(jù)到達(dá)時(shí)間對(duì)數(shù)據(jù)進(jìn)行分組。下面實(shí)驗(yàn)表明通過(guò)對(duì)會(huì)話進(jìn)行Clustering可以改善數(shù)據(jù)局部性并將查詢(xún)執(zhí)行時(shí)間減少50％以上。

查詢(xún)SQL如下

spark.sql("select  *  from table where session_id=123")

3.1 進(jìn)行Clustering之前

查詢(xún)花費(fèi)了2.2分鐘。請(qǐng)注意查詢(xún)計(jì)劃的"掃描parquet"部分中的輸出行數(shù)包括表中的所有2000W行。

3.2 進(jìn)行Clustering之后

查詢(xún)計(jì)劃與上面類(lèi)似，但由于改進(jìn)了數(shù)據(jù)局部性和謂詞下推，Spark可以修剪很多行。進(jìn)行Clustering后，相同的查詢(xún)?cè)趻呙鑠arquet文件時(shí)僅輸出11萬(wàn)行（2000萬(wàn)行中的），這將查詢(xún)時(shí)間從2.2分鐘減少到不到一分鐘。

下表總結(jié)了使用Spark3運(yùn)行的實(shí)驗(yàn)對(duì)查詢(xún)性能的改進(jìn)

Clustering后查詢(xún)運(yùn)行時(shí)間減少了60％，在其他樣本數(shù)據(jù)集上也觀察到了類(lèi)似的結(jié)果，請(qǐng)參閱示例查詢(xún)計(jì)劃和RFC-19性能評(píng)估上的更多詳細(xì)信息。

我們希望大型表能夠大幅度提高速度，與上面的示例不同，查詢(xún)運(yùn)行時(shí)間幾乎完全由實(shí)際I/O而不是查詢(xún)計(jì)劃決定。

4. 總結(jié)

使用Clustering，我們可以通過(guò)以下方式提高查詢(xún)性能：

利用空間填充曲線之類(lèi)的概念來(lái)適應(yīng)數(shù)據(jù)湖布局并減少查詢(xún)讀取的數(shù)據(jù)量。

將小文件合并成較大的文件以減少查詢(xún)引擎需要掃描的文件總數(shù)。

Clustering使得大數(shù)據(jù)進(jìn)行流處理，攝取可以寫(xiě)入小文件以滿足流處理的延遲要求，可以在后臺(tái)使用Clustering將這些小文件重寫(xiě)成較大的文件并減少文件數(shù)。

除此之外，Clustering框架還提供了根據(jù)特定要求異步重寫(xiě)數(shù)據(jù)的靈活性，我們預(yù)見(jiàn)到許多其他用例將采用帶有自定義可插拔策略的Clustering框架來(lái)按需管理數(shù)據(jù)湖數(shù)據(jù)，如可以通過(guò)Clustering解決如下一些用例：

重寫(xiě)數(shù)據(jù)并加密數(shù)據(jù)。

從表中修剪未使用的列并減少存儲(chǔ)空間。

以上就是Apache Hudi數(shù)據(jù)布局黑科技降低一半查詢(xún)時(shí)間的詳細(xì)內(nèi)容，更多關(guān)于Apache Hudi數(shù)據(jù)布局查詢(xún)的資料請(qǐng)關(guān)注本站其它相關(guān)文章！

國(guó)外服務(wù)器租用

版權(quán)聲明：本站文章來(lái)源標(biāo)注為YINGSOO的內(nèi)容版權(quán)均為本站所有，歡迎引用、轉(zhuǎn)載，請(qǐng)保持原文完整并注明來(lái)源及原文鏈接。禁止復(fù)制或仿造本網(wǎng)站，禁止在非www.sddonglingsh.com所屬的服務(wù)器上建立鏡像，否則將依法追究法律責(zé)任。本站部分內(nèi)容來(lái)源于網(wǎng)友推薦、互聯(lián)網(wǎng)收集整理而來(lái)，僅供學(xué)習(xí)參考，不代表本站立場(chǎng)，如有內(nèi)容涉嫌侵權(quán)，請(qǐng)聯(lián)系alex-e#qq.com處理。