Kyuubi 實(shí)踐 | 如何優(yōu)化 Spark 小文件，Kyuubi 一步搞定！
Hive 表中太多的小文件會(huì)影響數(shù)據(jù)的查詢性能和效率，同時(shí)加大了 HDFS NameNode 的壓力。Hive (on MapReduce) 一般可以簡單的通過一些參數(shù)來控制小文件，而 Spark 中并沒有提供小文件合并的功能。下面我們來簡單了解一下 Spark 小文件問題，以及如何處理小文件。
1.1 環(huán)境
Kyuubi 版本：1.6.0-SNAPSHOT
Spark 版本：3.1.3、3.2.0
1.2 TPCDS 數(shù)據(jù)集
Kyuubi 中提供了一個(gè) TPCDS Spark Connector，可以通過配置 Catalog 的方式，在讀取時(shí)自動(dòng)生成 TPCDS 數(shù)據(jù)。
只需要將 kyuubi-spark-connector-tpcds_2.12-1.6.0-SNAPSHOT.jar 包放入 Spark jars 目錄中，并配置：
spark.sql.catalog.tpcds=org.apache.kyuubi.spark.connector.tpcds.TPCDSCatalog;
這樣我們就可以直接讀取 TPCDS 數(shù)據(jù)集：
use tpcds;
show databases;
use sf3000;
show tables;
select * from sf300.catalog_returns limit 10；
1.3 小文件產(chǎn)生
首先我們在 Hive 中創(chuàng)建一個(gè) sample.catalog_returns 表，用于寫入生成的 TPCDScatalog_returns 數(shù)據(jù)，并添加一個(gè) hash 字段作為分區(qū)。
我們先關(guān)閉 Kyuubi 的優(yōu)化，讀取 catalog_returns 數(shù)據(jù)并寫入 Hive：
Spark SQL 最終產(chǎn)生的文件數(shù)最多可能是最后一個(gè)寫入的 Stage 的 Task 數(shù)乘以動(dòng)態(tài)分區(qū)的數(shù)量。我們 可以看到由于讀取輸入表的 Task 數(shù)是 44 個(gè)，所以最終產(chǎn)生了 44 個(gè)文件，每個(gè)文件大小約 69 M。
1.4 改變分區(qū)數(shù)（Repartition）
由于寫入的文件數(shù)跟最終寫入 Stage 的 Task 數(shù)據(jù)有關(guān)，那么我們可以通過添加一個(gè) Repartition 操作， 來減少最終寫入的 task 數(shù)，從而控制小文件：
添加 REPARTITION(10) 后，會(huì)在讀取后做一個(gè) Repartition 操作，將 partition 數(shù)變成 10，所以最終 寫入的文件數(shù)變成 10 個(gè)。
1.5 Spark AQE 自動(dòng)合并小分區(qū)
Spark 3.0 以后引入了自適應(yīng)查詢優(yōu)化（Adaptive Query Execution, AQE），可以自動(dòng)合并較小的分區(qū)。
開啟 AQE，并通過添加 distribute by cast(rand() * 100 as int) 觸發(fā) Shuffle 操作：
默認(rèn) Shuffle 分區(qū)數(shù) spark.sql.shuffle.partitions=200 ，如果不開啟 AQE 會(huì)產(chǎn)生 200 個(gè)小文件， 開啟 AQE 后，會(huì)自動(dòng)合并小分區(qū)，根據(jù)spark.sql.adaptive.advisoryPartitionSizeInBytes=512M 配置合并較小的分區(qū)，最終產(chǎn)生 12 個(gè)文件。
Apache Kyuubi (Incubating) 作為增強(qiáng)版的 Spark Thrift Server 服務(wù)，可通過 Spark SQL 進(jìn)行大規(guī)模的 數(shù)據(jù)處理分析。Kyuubi 通過 Spark SQL Extensions 實(shí)現(xiàn)了很多的 Spark 優(yōu)化，其中包括了RepartitionBeforeWrite 的優(yōu)化，再結(jié)合 Spark AQE 可以自動(dòng)優(yōu)化小文件問題，下面我們具體分析 一下 Kyuubi 如何實(shí)現(xiàn)小文件優(yōu)化。
2.1 Kyuubi 如何優(yōu)化小文件
Kyuubi 提供了在寫入前加上 Repartition 操作的優(yōu)化，我們只需要將 kyuubi-extension-spark-3- 1_2.12-1.6.0-SNAPSHOT.jar 放入 Spark jars 目錄中，并配置spark.sql.extensions=org.apache.kyuubi.sql.KyuubiSparkSQLExtension 。相關(guān)配置：
通過 spark.sql.optimizer.insertRepartitionNum 參數(shù)可以配置最終插入 Repartition 的分區(qū)數(shù)， 當(dāng)不開啟 AQE，默認(rèn)為 spark.sql.shuffle.partitions 的值。需要注意，當(dāng)我們設(shè)置此配置會(huì)導(dǎo)致 AQE 失效，所以開啟 AQE 不建議設(shè)置此值。
對于動(dòng)態(tài)分區(qū)寫入，會(huì)根據(jù)動(dòng)態(tài)分區(qū)字段進(jìn)行 Repartition，并添加一個(gè)隨機(jī)數(shù)來避免產(chǎn)生數(shù)據(jù)傾斜，spark.sql.optimizer.dynamicPartitionInsertionRepartitionNum 用來配置隨機(jī)數(shù)的范圍，不 過添加隨機(jī)數(shù)后，由于加大了動(dòng)態(tài)分區(qū)的基數(shù)，還是可能會(huì)導(dǎo)致小文件。這個(gè)操作類似在 SQL 中添加distribute by DYNAMIC_PARTITION_COLUMN, cast(rand() * 100 as int)。
2.2 靜態(tài)分區(qū)寫入
開啟 Kyuubi 優(yōu)化和 AQE，測試靜態(tài)分區(qū)寫入：
可以看到 AQE 生效了，很好地控制了小文件，產(chǎn)生了 11 個(gè)文件，文件大小 314.5 M 左右。
2.3 動(dòng)態(tài)分區(qū)寫入
我們測試一下動(dòng)態(tài)分區(qū)寫入的情況，先關(guān)閉 Kyuubi 優(yōu)化，并生成 10 個(gè) hash 分區(qū)：
產(chǎn)生了 44 × 10 = 440 個(gè)文件，文件大小 8 M 左右。
開啟 Kyuubi 優(yōu)化和 AQE：
產(chǎn)生了 12 × 10 = 120 個(gè)文件，文件大小 30 M 左右，可以看到小文件有所改善，不過仍然不夠理想。
此案例中 hash 分區(qū)由 rand 函數(shù)產(chǎn)生，分布比較均勻，所以我們將spark.sql.optimizer.dynamicPartitionInsertionRepartitionNum 設(shè)置成 0 ，重新運(yùn)行，同時(shí)將動(dòng)態(tài)分區(qū)數(shù)設(shè)置為5 ：
由于動(dòng)態(tài)分區(qū)數(shù)只有 5 個(gè)，所以實(shí)際上只有 5 個(gè) Task 有數(shù)據(jù)寫入，每個(gè) Task 對應(yīng)一個(gè)分區(qū)，導(dǎo)致最終每個(gè)分區(qū)只有一個(gè)較大的大文件。
通過上面的分析可以看到，對于動(dòng)態(tài)分區(qū)寫入，Repartition 的優(yōu)化可以緩解小文件，配置spark.sql.optimizer.dynamicPartitionInsertionRepartitionNum=100 解決了數(shù)據(jù)傾斜問題， 不過同時(shí)還是可能會(huì)有小文件。
2.4 Rebalance 優(yōu)化
Spark 3.2+ 引入了 Rebalance 操作，借助于 Spark AQE 來平衡分區(qū)，進(jìn)行小分區(qū)合并和傾斜分區(qū)拆 分，避免分區(qū)數(shù)據(jù)過大或過小，能夠很好地處理小文件問題。
Kyuubi 對于 Spark 3.2+ 的優(yōu)化，是在寫入前插入 Rebalance 操作，對于動(dòng)態(tài)分區(qū)，則指定動(dòng)態(tài)分區(qū)列 進(jìn)行 Rebalance 操作。不再需要 spark.sql.optimizer.insertRepartitionNum 和spark.sql.optimizer.dynamicPartitionInsertionRepartitionNum 配置。
測試靜態(tài)分區(qū)寫入，使用 Spark 3.2.0 開啟 Kyuubi 優(yōu)化和 AQE：
Repartition 操作自動(dòng)合并了小分區(qū)，產(chǎn)生了 11 個(gè)文件，文件大小 334.6 M 左右，解決了小文件的問題。
測試動(dòng)態(tài)分區(qū)寫入，使用 Spark 3.2.0 開啟 Kyuubi 優(yōu)化和 AQE，生成 5 個(gè)動(dòng)態(tài)分區(qū)：
Repartition 操作自動(dòng)拆分較大分區(qū)，產(chǎn)生了 2 × 5 = 10 個(gè)文件，文件大小 311 M 左右，很好地解決傾斜問題。
從上面的分析可以看到，對于 Spark 3.2+，Kyuubi 結(jié)合 Rebalance 能夠很好地解決小文件問題，對于 Spark 3.1，Kyuubi 也能自動(dòng)優(yōu)化小文件，不過動(dòng)態(tài)分區(qū)寫入的情況還是可能存在問題。
相關(guān)的配置總結(jié)：
更多 AQE 配置可以參考：How To Use Spark Adaptive Query Execution (AQE) in Kyuubi
END
如果你也想成為 Committer
????????????
·Become A Committer of Apache Kyuubi
