Apache Hudi性能提升三倍的查詢優(yōu)化
從 Hudi 0.10.0版本開始，我們很高興推出在數(shù)據(jù)庫領(lǐng)域中稱為 Z-Order 和 Hilbert 空間填充曲線的高級數(shù)據(jù)布局優(yōu)化技術(shù)的支持。
1. 背景
Amazon EMR 團(tuán)隊最近發(fā)表了一篇很不錯的文章展示了對數(shù)據(jù)進(jìn)行聚簇是如何提高查詢性能的，為了更好地了解發(fā)生了什么以及它與空間填充曲線的關(guān)系，讓我們仔細(xì)研究該文章的設(shè)置。
文章中比較了 2 個 Apache Hudi 表（均來自 Amazon Reviews 數(shù)據(jù)集）：
未聚簇的 amazon_reviews 表（即數(shù)據(jù)尚未按任何特定鍵重新排序）
amazon_reviews_clustered 聚簇表。當(dāng)數(shù)據(jù)被聚簇后，數(shù)據(jù)按字典順序排列（這里我們將這種排序稱為線性排序），排序列為star_rating、total_votes兩列（見下圖）
為了展示查詢性能的改進(jìn)，對這兩個表執(zhí)行以下查詢：
這里要指出的重要考慮因素是查詢指定了排序的兩個列（star_rating 和 total_votes）。但不幸的是這是線性/詞典排序的一個關(guān)鍵限制，如果添加更多列，排序的價值會會隨之減少。
從上圖可以看到，對于按字典順序排列的 3 元組整數(shù)，只有第一列能夠?qū)λ芯哂邢嗤档挠涗浘哂嘘P(guān)鍵的局部性屬性：例如所有記錄都具有以“開頭的值” 1"、"2"、"3"（在第一列中）很好地聚簇在一起。但是如果嘗試在第三列中查找所有值為"5"的值，會發(fā)現(xiàn)這些值現(xiàn)在分散在所有地方，根本沒有局部性，過濾效果很差。
提高查詢性能的關(guān)鍵因素是局部性：它使查詢能夠顯著減少搜索空間和需要掃描、解析等的文件數(shù)量。
但是這是否意味著如果我們按表排序的列的第一個（或更準(zhǔn)確地說是前綴）以外的任何內(nèi)容進(jìn)行過濾，我們的查詢就注定要進(jìn)行全面掃描？不完全是，局部性也是空間填充曲線在枚舉多維空間時啟用的屬性（我們表中的記錄可以表示為 N 維空間中的點，其中 N 是我們表中的列數(shù)）
那么它是如何工作的？我們以 Z 曲線為例：擬合二維平面的 Z 階曲線如下所示：
可以看到按照路徑，不是簡單地先按一個坐標(biāo) ("x") 排序，然后再按另一個坐標(biāo)排序，它實際上是在對它們進(jìn)行排序，就好像這些坐標(biāo)的位已交織成單個值一樣：
在線性排序的情況下局部性僅使用第一列相比，該方法的局部性使用到所有列。
以類似的方式，希爾伯特曲線允許將 N 維空間中的點（我們表中的行）映射到一維曲線上，基本上對它們進(jìn)行排序，同時仍然保留局部性的關(guān)鍵屬性，在此處閱讀有關(guān)希爾伯特曲線的更多詳細(xì)信息，到目前為止我們的實驗表明，使用希爾伯特曲線對數(shù)據(jù)進(jìn)行排序會有更好的聚簇和性能結(jié)果。
現(xiàn)在讓我們來看看它的實際效果！
2. 設(shè)置
我們將再次使用 Amazon Reviews 數(shù)據(jù)集，但這次我們將使用 Hudi 按 product_id、customer_id 列元組進(jìn)行 Z-Order排序，而不是聚簇或線性排序。
數(shù)據(jù)集不需要特別的準(zhǔn)備，可以直接從 S3 中以 Parquet 格式下載并將其直接用作 Spark 將其攝取到 Hudi 表。
啟動spark-shell
./bin/spark-shell --master 'local[4]' --driver-memory 8G --executor-memory 8G \ --jars ../../packaging/hudi-spark-bundle/target/hudi-spark3-bundle_2.12-0.10.0.jar \ --packages org.apache.spark:spark-avro_2.12:2.4.4 \ --conf 'spark.serializer=org.apache.spark.serializer.KryoSerializer'
導(dǎo)入Hudi表
import org.apache.hadoop.fs.{FileStatus, Path}
import scala.collection.JavaConversions._
import org.apache.spark.sql.SaveMode._
import org.apache.hudi.{DataSourceReadOptions, DataSourceWriteOptions}
import org.apache.hudi.DataSourceWriteOptions._
import org.apache.hudi.common.fs.FSUtils
import org.apache.hudi.common.table.HoodieTableMetaClient
import org.apache.hudi.common.util.ClusteringUtils
import org.apache.hudi.config.HoodieClusteringConfig
import org.apache.hudi.config.HoodieWriteConfig._
import org.apache.spark.sql.DataFrame
import java.util.stream.Collectors
val layoutOptStrategy = "z-order"; // OR "hilbert"
val inputPath = s"file:///${System.getProperty("user.home")}/datasets/amazon_reviews_parquet"
val tableName = s"amazon_reviews_${layoutOptStrategy}"
val outputPath = s"file:///tmp/hudi/$tableName"
def safeTableName(s: String) = s.replace('-', '_')
val commonOpts =
Map(
"hoodie.compact.inline" -> "false",
"hoodie.bulk_insert.shuffle.parallelism" -> "10"
)
////////////////////////////////////////////////////////////////
// Writing to Hudi
////////////////////////////////////////////////////////////////
val df = spark.read.parquet(inputPath)
df.write.format("hudi")
.option(DataSourceWriteOptions.TABLE_TYPE.key(), COW_TABLE_TYPE_OPT_VAL)
.option("hoodie.table.name", tableName)
.option(PRECOMBINE_FIELD.key(), "review_id")
.option(RECORDKEY_FIELD.key(), "review_id")
.option(DataSourceWriteOptions.PARTITIONPATH_FIELD.key(), "product_category")
.option("hoodie.clustering.inline", "true")
.option("hoodie.clustering.inline.max.commits", "1")
// NOTE: Small file limit is intentionally kept _ABOVE_ target file-size max threshold for Clustering,
// to force re-clustering
.option("hoodie.clustering.plan.strategy.small.file.limit", String.valueOf(1024 * 1024 * 1024)) // 1Gb
.option("hoodie.clustering.plan.strategy.target.file.max.bytes", String.valueOf(128 * 1024 * 1024)) // 128Mb
// NOTE: We're increasing cap on number of file-groups produced as part of the Clustering run to be able to accommodate for the
// whole dataset (~33Gb)
.option("hoodie.clustering.plan.strategy.max.num.groups", String.valueOf(4096))
.option(HoodieClusteringConfig.LAYOUT_OPTIMIZE_ENABLE.key, "true")
.option(HoodieClusteringConfig.LAYOUT_OPTIMIZE_STRATEGY.key, layoutOptStrategy)
.option(HoodieClusteringConfig.PLAN_STRATEGY_SORT_COLUMNS.key, "product_id,customer_id")
.option(DataSourceWriteOptions.OPERATION.key(), DataSourceWriteOptions.BULK_INSERT_OPERATION_OPT_VAL)
.option(BULK_INSERT_SORT_MODE.key(), "NONE")
.options(commonOpts)
.mode(ErrorIfExists)
3. 測試
每個單獨(dú)的測試請在單獨(dú)的 spark-shell 中運(yùn)行，以避免緩存影響測試結(jié)果。
////////////////////////////////////////////////////////////////
// Reading
///////////////////////////////////////////////////////////////
// Temp Table w/ Data Skipping DISABLED
val readDf: DataFrame =
spark.read.option(DataSourceReadOptions.ENABLE_DATA_SKIPPING.key(), "false").format("hudi").load(outputPath)
val rawSnapshotTableName = safeTableName(s"${tableName}_sql_snapshot")
readDf.createOrReplaceTempView(rawSnapshotTableName)
// Temp Table w/ Data Skipping ENABLED
val readDfSkip: DataFrame =
spark.read.option(DataSourceReadOptions.ENABLE_DATA_SKIPPING.key(), "true").format("hudi").load(outputPath)
val dataSkippingSnapshotTableName = safeTableName(s"${tableName}_sql_snapshot_skipping")
readDfSkip.createOrReplaceTempView(dataSkippingSnapshotTableName)
// Query 1: Total votes by product_category, for 6 months
def runQuery1(tableName: String) = {
// Query 1: Total votes by product_category, for 6 months
spark.sql(s"SELECT sum(total_votes), product_category FROM $tableName WHERE review_date > '2013-12-15' AND review_date < '2014-06-01' GROUP BY product_category").show()
}
// Query 2: Average star rating by product_id, for some product
def runQuery2(tableName: String) = {
spark.sql(s"SELECT avg(star_rating), product_id FROM $tableName WHERE product_id in ('B0184XC75U') GROUP BY product_id").show()
}
// Query 3: Count number of reviews by customer_id for some 5 customers
def runQuery3(tableName: String) = {
spark.sql(s"SELECT count(*) as num_reviews, customer_id FROM $tableName WHERE customer_id in ('53096570','10046284','53096576','10000196','21700145') GROUP BY customer_id").show()
}
//
// Query 1: Is a "wide" query and hence it's expected to touch a lot of files
//
scala> runQuery1(rawSnapshotTableName)
+----------------+--------------------+
|sum(total_votes)| product_category|
+----------------+--------------------+
| 1050944| PC|
| 867794| Kitchen|
| 1167489| Home|
| 927531| Wireless|
| 6861| Video|
| 39602| Digital_Video_Games|
| 954924|Digital_Video_Dow...|
| 81876| Luggage|
| 320536| Video_Games|
| 817679| Sports|
| 11451| Mobile_Electronics|
| 228739| Home_Entertainment|
| 3769269|Digital_Ebook_Pur...|
| 252273| Baby|
| 735042| Apparel|
| 49101| Major_Appliances|
| 484732| Grocery|
| 285682| Tools|
| 459980| Electronics|
| 454258| Outdoors|
+----------------+--------------------+
only showing top 20 rows
scala> runQuery1(dataSkippingSnapshotTableName)
+----------------+--------------------+
|sum(total_votes)| product_category|
+----------------+--------------------+
| 1050944| PC|
| 867794| Kitchen|
| 1167489| Home|
| 927531| Wireless|
| 6861| Video|
| 39602| Digital_Video_Games|
| 954924|Digital_Video_Dow...|
| 81876| Luggage|
| 320536| Video_Games|
| 817679| Sports|
| 11451| Mobile_Electronics|
| 228739| Home_Entertainment|
| 3769269|Digital_Ebook_Pur...|
| 252273| Baby|
| 735042| Apparel|
| 49101| Major_Appliances|
| 484732| Grocery|
| 285682| Tools|
| 459980| Electronics|
| 454258| Outdoors|
+----------------+--------------------+
only showing top 20 rows
//
// Query 2: Is a "pointwise" query and hence it's expected that data-skipping should substantially reduce number
// of files scanned (as compared to Baseline)
//
// NOTE: That Linear Ordering (as compared to Space-curve based on) will have similar effect on performance reducing
// total # of Parquet files scanned, since we're querying on the prefix of the ordering key
//
scala> runQuery2(rawSnapshotTableName)
+----------------+----------+
|avg(star_rating)|product_id|
+----------------+----------+
| 1.0|B0184XC75U|
+----------------+----------+
scala> runQuery2(dataSkippingSnapshotTableName)
+----------------+----------+
|avg(star_rating)|product_id|
+----------------+----------+
| 1.0|B0184XC75U|
+----------------+----------+
//
// Query 3: Similar to Q2, is a "pointwise" query, but querying other part of the ordering-key (product_id, customer_id)
// and hence it's expected that data-skipping should substantially reduce number of files scanned (as compared to Baseline, Linear Ordering).
//
// NOTE: That Linear Ordering (as compared to Space-curve based on) will _NOT_ have similar effect on performance reducing
// total # of Parquet files scanned, since we're NOT querying on the prefix of the ordering key
//
scala> runQuery3(rawSnapshotTableName)
+-----------+-----------+
|num_reviews|customer_id|
+-----------+-----------+
| 50| 53096570|
| 3| 53096576|
| 25| 10046284|
| 1| 10000196|
| 14| 21700145|
+-----------+-----------+
scala> runQuery3(dataSkippingSnapshotTableName)
+-----------+-----------+
|num_reviews|customer_id|
+-----------+-----------+
| 50| 53096570|
| 3| 53096576|
| 25| 10046284|
| 1| 10000196|
| 14| 21700145|
+-----------+-----------+
4. 結(jié)果
我們總結(jié)了以下的測試結(jié)果
可以看到多列線性排序?qū)τ诎戳校≦2、Q3）以外的列進(jìn)行過濾的查詢不是很有效，這與空間填充曲線（Z-order 和 Hilbert）形成了非常明顯的對比，后者將查詢時間加快多達(dá) 3倍 。值得注意的是性能提升在很大程度上取決于基礎(chǔ)數(shù)據(jù)和查詢，在我們內(nèi)部數(shù)據(jù)的基準(zhǔn)測試中，能夠?qū)崿F(xiàn)超過 11倍 的查詢性能改進(jìn)！
5. 總結(jié)
Apache Hudi v0.10 為開源帶來了新的布局優(yōu)化功能 Z-order 和 Hilbert。 使用這些行業(yè)領(lǐng)先的布局優(yōu)化技術(shù)可以為用戶查詢帶來顯著的性能提升和成本節(jié)約！
以上就是Apache Hudi性能提升三倍的查詢優(yōu)化的詳細(xì)內(nèi)容，更多關(guān)于Apache Hudi查詢優(yōu)化的資料請關(guān)注腳本之家其它相關(guān)文章！
相關(guān)文章
網(wǎng)絡(luò)線路科普之CN2，GIA，CIA，BGP以及IPLC都是什么意思
購買海外vps或者物理服務(wù)器或者海外實體服務(wù)器托管的時候,在中國IDC服務(wù)器商中的有關(guān)網(wǎng)絡(luò)線路帶寬的術(shù)語有很多,今天專門做了一個專題,有關(guān)IPLC專線、CN2、BGP、CIA和普通線路知識普及2021-06-06
GitHub是一個免費(fèi)托管開源代碼的遠(yuǎn)程倉庫,但當(dāng)我們既不想公開源代碼,又舍不得給GitHub交保護(hù)費(fèi)時,就需要自己搭建一臺Git服務(wù)器作為私有倉庫使用了,下面我們就來看看具體實現(xiàn)方法吧2023-08-08
這篇文章主要介紹了云服務(wù)器搭建Hadoop分布式的過程,本文給大家介紹的非常詳細(xì),對大家的學(xué)習(xí)或工作具有一定的參考借鑒價值,需要的朋友可以參考下2023-10-10
IBM X346 板載 Raid1 功能啟用及Win2003系統(tǒng)安裝小記
有兩臺IBM X346 m/t-8840-cdr 服務(wù)器，板載raid控制器版本號為v.4.30.9，硬盤位0，1各裝一塊146G IBM 原裝SCSI硬盤。2011-04-04
TCP服務(wù)器實現(xiàn)數(shù)據(jù)通信
今天我們要介紹的是使用TCP協(xié)議實現(xiàn)數(shù)據(jù)通信，相比于之前寫的UDP服務(wù)器實現(xiàn)數(shù)據(jù)信，在主體邏輯上并沒有差別，下面，我們具體來看看UDP和TCP在編碼的實現(xiàn)上有什么不同，感興趣的朋友一起看看吧2023-08-08
360提示服務(wù)器開啟了WebDAV的關(guān)閉方法
WebDAV （Web-based Distributed Authoring and Versioning） 一種基于 HTTP 1.1協(xié)議的通信協(xié)議.它擴(kuò)展了HTTP 1.1，在GET、POST、HEAD等幾個HTTP標(biāo)準(zhǔn)方法以外添加了一些新的方法2012-07-07
HTTPS并不是一種新技術(shù)，它是在HTTP協(xié)議的基礎(chǔ)上來進(jìn)行更嚴(yán)格的加密。這篇文章主要介紹了https協(xié)議詳解的相關(guān)資料,需要的朋友可以參考下2022-10-10
Kubernetes特別屬性的標(biāo)簽Label的強(qiáng)大作用
這篇文章主要為大家介紹了Kubernetes特別屬性的標(biāo)簽Label的強(qiáng)大作用介紹，希望可以對大家有所幫助，祝大家多多進(jìn)步，早日升職加薪2022-03-03
近期服務(wù)器出現(xiàn)的安全問題以及防范措施2017.05
近期接到idc商的反饋，最近很多使用windows的主機(jī)都被拿下控制權(quán)，直接修改iis等2017-08-08
git標(biāo)簽管理_動力節(jié)點Java學(xué)院整理
這篇文章主要為大家詳細(xì)介紹了git標(biāo)簽管理的相關(guān)資料，具有一定的參考價值，感興趣的小伙伴們可以參考一下2017-08-08