1. 背景

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

2. Clustering架構(gòu)

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

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

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

總體而言Clustering分為兩個部分：

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

2.1 調(diào)度Clustering

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

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

2.2 運行Clustering

•讀取Clustering計劃，并獲得clusteringGroups，其標(biāo)記了需要進行Clustering的文件組。•對于每個組使用strategyParams實例化適當(dāng)?shù)牟呗灶悾ɡ纾簊ortColumns），然后應(yīng)用該策略重寫數(shù)據(jù)。•創(chuàng)建一個REPLACE提交，并更新HoodieReplaceCommitMetadata中的元數(shù)據(jù)。

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

注意：現(xiàn)在對表進行Clustering時還不支持更新，將來會支持并發(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");

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

3. 表查詢性能

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

查詢SQL如下

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

3.1 進行Clustering之前

查詢花費了2.2分鐘。請注意查詢計劃的"掃描parquet"部分中的輸出行數(shù)包括表中的所有2000W行。

3.2 進行Clustering之后

查詢計劃與上面類似，但由于改進了數(shù)據(jù)局部性和謂詞下推，Spark可以修剪很多行。進行Clustering后，相同的查詢在掃描parquet文件時僅輸出11萬行（2000萬行中的），這將查詢時間從2.2分鐘減少到不到一分鐘。

下表總結(jié)了使用Spark3運行的實驗對查詢性能的改進

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

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

4. 總結(jié)

使用Clustering，我們可以通過以下方式提高查詢性能：

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

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

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

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

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

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

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

最新評論