查詢效能提升3倍！Apache Hudi 查詢優化了解下？
阿新 • • 發佈：2022-03-06
從 Hudi 0.10.0版本開始，我們很高興推出在資料庫領域中稱為 Z-Order 和 Hilbert 空間填充曲線的高階資料佈局優化技術的支援。
1. 背景
Amazon EMR 團隊最近發表了一篇很不錯的文章展示了對資料進行聚簇是如何提高查詢效能的，為了更好地瞭解發生了什麼以及它與空間填充曲線的關係，讓我們仔細研究該文章的設定。
文章中比較了 2 個 Apache Hudi 表（均來自 Amazon Reviews 資料集）：
-
未聚簇的 amazon_reviews 表（即資料尚未按任何特定鍵重新排序）
-
amazon_reviews_clustered 聚簇表。當資料被聚簇後，資料按字典順序排列（這裡我們將這種排序稱為線性排序），排序列為
star_rating、total_votes兩列（見下圖）
為了展示查詢效能的改進，對這兩個表執行以下查詢：
這裡要指出的重要考慮因素是查詢指定了排序的兩個列（star_rating 和 total_votes）。但不幸的是這是線性/詞典排序的一個關鍵限制，如果新增更多列，排序的價值會會隨之減少。
從上圖可以看到，對於按字典順序排列的 3 元組整數，只有第一列能夠對所有具有相同值的記錄具有關鍵的區域性性屬性：例如所有記錄都具有以“開頭的值” 1"、"2"、"3"（在第一列中）很好地聚簇在一起。但是如果嘗試在第三列中查詢所有值為"5"的值，會發現這些值現在分散在所有地方，根本沒有區域性性，過濾效果很差。
提高查詢效能的關鍵因素是區域性性：它使查詢能夠顯著減少搜尋空間和需要掃描、解析等的檔案數量。
但是這是否意味著如果我們按表排序的列的第一個（或更準確地說是字首）以外的任何內容進行過濾，我們的查詢就註定要進行全面掃描？不完全是，區域性性也是空間填充曲線在列舉多維空間時啟用的屬性（我們表中的記錄可以表示為 N 維空間中的點，其中 N 是我們表中的列數）
那麼它是如何工作的？我們以 Z 曲線為例：擬合二維平面的 Z 階曲線如下所示：
可以看到按照路徑，不是簡單地先按一個座標 ("x") 排序，然後再按另一個座標排序，它實際上是在對它們進行排序，就好像這些座標的位已交織成單個值一樣：
線上性排序的情況下區域性性僅使用第一列相比，該方法的區域性性使用到所有列。
以類似的方式，希爾伯特曲線允許將 N 維空間中的點（我們表中的行）對映到一維曲線上，基本上對它們進行排序，同時仍然保留區域性性的關鍵屬性，在 此處閱讀有關希爾伯特曲線的更多詳細資訊，到目前為止我們的實驗表明，使用希爾伯特曲線對資料進行排序會有更好的聚簇和效能結果。
現在讓我們來看看它的實際效果！
2. 設定
我們將再次使用 Amazon Reviews 資料集，但這次我們將使用 Hudi 按 product_id、customer_id 列元組進行 Z-Order排序，而不是聚簇或線性排序。
資料集不需要特別的準備，可以直接從 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'
匯入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. 測試
每個單獨的測試請在單獨的 spark-shell 中執行，以避免快取影響測試結果。
////////////////////////////////////////////////////////////////
// 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. 結果
我們總結了以下的測試結果
可以看到多列線性排序對於按列（Q2、Q3）以外的列進行過濾的查詢不是很有效，這與空間填充曲線（Z-order 和 Hilbert）形成了非常明顯的對比，後者將查詢時間加快多達 3倍 。值得注意的是效能提升在很大程度上取決於基礎資料和查詢，在我們內部資料的基準測試中，能夠實現超過 11倍 的查詢效能改進！
5. 總結
Apache Hudi v0.10 為開源帶來了新的佈局優化功能 Z-order 和 Hilbert。 使用這些行業領先的佈局優化技術可以為使用者查詢帶來顯著的效能提升和成本節約！
PS:如果您覺得閱讀本文對您有幫助，請點一下“推薦”按鈕，您的“推薦”，將會是我不竭的動力！
作者：leesf掌控之中，才會成功；掌控之外，註定失敗。
出處：http://www.cnblogs.com/leesf456/
本文版權歸作者和部落格園共有，歡迎轉載，但未經作者同意必須保留此段宣告，且在文章頁面明顯位置給出原文連線，否則保留追究法律責任的權利。
如果覺得本文對您有幫助，您可以請我喝杯咖啡!