动态分区

使用案例：

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动态申请

使用案例：

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

ORC文件性能优化

使用案例：

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;

补充说明：

在决定一个目录是否需要合并小文件时，会统计目录下的平均大小，然后和spark.sql.mergeSmallFileSize比较
在合并文件时，一个map task读取的数据量取决于下面三者的较大值：
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. 自动设置Shuffle Partition个数功能已经发布，而动态调整执行计划和自动处理数据倾斜 还不确定是否已经发布（Spark Adaptive Execution调研）
2. 自动设置Shuffle Partition功能（spark.sql.adaptive.shuffle.targetPostShuffleInputSize）慎用，可能出现极端情况而导致耗时变长。比如，笔者有次在一个任务中使用该功能，当时读取表数据量是千亿级别的，经过shuffle处理（broadcast join）后最终的结果数据量在万级别，大小在几十kb，因此shuffle后的分区数变成了1，但是shuffle前的数据量是很大的，因此分区数也达到了笔者设置的上限5000。也就是说，最后是一个task去拉取5000个task写好的shuffle文件，虽然这些shuffle文件大多很小，甚至是没有，但是整个拉取过程依然是非常耗时。而我之所以用这个功能的原因是只是为了小文件合并，完全没有必要用这个参数，将这个功能参数注释掉，换成小文件合并相关的参数后，任务执行时长就下降了十几分钟
3. 自动处理数据倾斜原理：根据分区数据大小和条数来判断该分区是否为倾斜分区，如果是倾斜分区，就会将该分区对应task由1个改为多个，而另一部分的相应partition数据全量传输。比如，表a的partition0有倾斜情况，表b的partition0没有数据倾斜情况，此时会将表a的partition0分为多个部分传到不同的core上处理，同时表b的partition0数据也会全量shuffle到这些core上，因此表b的partition0最好不要数据量过大（Adaptive Execution让Spark SQL更高效更智能）

推测执行

使用案例：

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