動態(tài)分區(qū)

使用案例：

set hive.exec.dynamic.partition=true;
set hive.exec.dynamic.partition.mode=nonstrict;
set hive.exec.max.dynamic.partitions=5000;

資源申請

使用案例：

set spark.executor.cores=2;
set spark.executor.memory=4G;
set spark.yarn.executor.memoryOverhead=1024;
set spark.driver.memory=8G;
set spark.yarn.driver.memoryOverhead=1024;
set spark.memory.fraction=0.7;

Executor動態(tài)申請

使用案例：

set spark.dynamicAllocation.enabled=true;
set spark.dynamicAllocation.maxExecutors=1000;
set spark.dynamicAllocation.minExecutors=400;

ORC文件性能優(yōu)化

使用案例：

set spark.sql.orc.filterPushdown=true;
set spark.sql.orc.splits.include.file.footer=true;
set spark.sql.orc.cache.stripe.details.size=10000;
set spark.sql.hive.metastorePartitionPruning=true;

補充說明：ORC文件存儲格式如下表

ORC文件存儲格式.png

文件輸入輸出

使用案例：

set spark.hadoop.hive.exec.orc.split.strategy=ETL;
set spark.hadoop.mapreduce.input.fileinputformat.split.minsize=67108864;
set spark.hadoop.mapreduce.input.fileinputformat.split.maxsize=268435456;
set spark.hadoop.mapreduce.fileoutputcommitter.algorithm.version=2;

小文件合并

使用案例

set spark.sql.mergeSmallFileSize=67108864;
set spark.sql.targetBytesInPartitionWhenMerge=67108864;
set spark.hadoopRDD.targetBytesInPartition=67108864;

補充說明：

在決定一個目錄是否需要合并小文件時，會統(tǒng)計目錄下的平均大小，然后和spark.sql.mergeSmallFileSize比較
在合并文件時，一個map task讀取的數(shù)據(jù)量取決于下面三者的較大值：
1、spark.sql.mergeSmallFileSize
2、spark.sql.targetBytesInPartitionWhenMerge
3、spark.hadoopRDD.targetBytesInPartition

Shuffle

使用案例：

set spark.sql.autoBroadcastJoinThreshold=33554432;
set spark.sql.shuffle.partitions=5000;
set spark.reducer.maxSizeInFlight=25165824;
set spark.reducer.maxReqsInFlight=10;
set spark.reducer.maxBlocksInFlightPerAddress=1;
set spark.reducer.maxReqSizeShuffleToMem=536870911;
set spark.shuffle.io.connectionTimeout=120;
set spark.shuffle.io.maxRetries=3;
set spark.shuffle.io.retryWait=5;

Adaptive Execution

使用案例：

set spark.sql.adaptive.enabled=true;
set spark.sql.adaptive.shuffle.targetPostShuffleInputSize=268435456;
set spark.sql.adaptive.join.enabled=true;
set spark.sql.adaptiveBroadcastJoinThreshold=33554432;
set spark.sql.adaptive.allowAddititionalShuffle=false;
set spark.sql.adaptive.skewedJoin.enabled=true;
set spark.sql.adaptive.skewedPartitionMaxSplits=100;
set spark.sql.adaptive.skewedPartitionRowCountThreshold=10000000;
set spark.sql.adaptive.skewedPartitionSizeThreshold=536870912;
set spark.sql.adaptive.skewedPartitionFactor=10;

補充說明：

1. 自動設(shè)置Shuffle Partition個數(shù)功能已經(jīng)發(fā)布，而動態(tài)調(diào)整執(zhí)行計劃和自動處理數(shù)據(jù)傾斜 還不確定是否已經(jīng)發(fā)布（Spark Adaptive Execution調(diào)研）
2. 自動設(shè)置Shuffle Partition功能（spark.sql.adaptive.shuffle.targetPostShuffleInputSize）慎用，可能出現(xiàn)極端情況而導(dǎo)致耗時變長。比如，筆者有次在一個任務(wù)中使用該功能，當(dāng)時讀取表數(shù)據(jù)量是千億級別的，經(jīng)過shuffle處理（broadcast join）后最終的結(jié)果數(shù)據(jù)量在萬級別，大小在幾十kb，因此shuffle后的分區(qū)數(shù)變成了1，但是shuffle前的數(shù)據(jù)量是很大的，因此分區(qū)數(shù)也達(dá)到了筆者設(shè)置的上限5000。也就是說，最后是一個task去拉取5000個task寫好的shuffle文件，雖然這些shuffle文件大多很小，甚至是沒有，但是整個拉取過程依然是非常耗時。而我之所以用這個功能的原因是只是為了小文件合并，完全沒有必要用這個參數(shù)，將這個功能參數(shù)注釋掉，換成小文件合并相關(guān)的參數(shù)后，任務(wù)執(zhí)行時長就下降了十幾分鐘
3. 自動處理數(shù)據(jù)傾斜原理：根據(jù)分區(qū)數(shù)據(jù)大小和條數(shù)來判斷該分區(qū)是否為傾斜分區(qū)，如果是傾斜分區(qū)，就會將該分區(qū)對應(yīng)task由1個改為多個，而另一部分的相應(yīng)partition數(shù)據(jù)全量傳輸。比如，表a的partition0有傾斜情況，表b的partition0沒有數(shù)據(jù)傾斜情況，此時會將表a的partition0分為多個部分傳到不同的core上處理，同時表b的partition0數(shù)據(jù)也會全量shuffle到這些core上，因此表b的partition0最好不要數(shù)據(jù)量過大（Adaptive Execution讓Spark SQL更高效更智能）

推測執(zhí)行

使用案例：

set spark.speculation=true;
set spark.speculation.interval=1000ms;
set spark.speculation.quantile=0.99;
set spark.speculation.multiplier=3;