spark 任务优化

数据倾斜

处理方法

1.过滤掉不符合预期的热点key，例如由于日志信息丢失导致某个字段产生大量空值
2.加入随机因素，打散热点key
3.使用map join解决小表关联大表造成的数据倾斜问题

文件

大量的小文件会影响Hadoop集群管理或者Spark在处理数据时的稳定性：

处理方法

向量化读取

spark.sql.parquet.enableVectorizedReader=true

调整输入文件分割

– 小文件合并的阈值
set spark.sql.mergeSmallFileSize=268435456;
– 小文件合并的task中，每个task读取的数据量
set spark.sql.targetBytesInPartitionWhenMerge=268435456;
set spark.hadoopRDD.targetBytesInPartition=268435456;
set spark.sql.files.maxPartitionBytes=268435456;
文本用spark.mapreduce.input.fileinputformat.split.minsize

shuffle

如果数据源shuffle产生，依赖于spark.sql.shuffle.partitions配置信息，默认为200，降低spark.sql.shuffle.partitions配置的值，但会影响处理的并发度

自适应

打开自适应框架的开关
spark.sql.adaptive.enabled true
设置partition的上下限
spark.sql.adaptive.minNumPostShufflePartitions 10
spark.sql.adaptive.maxNumPostShufflePartitions 2000
设置单reduce task处理的数据大小
spark.sql.adaptive.shuffle.targetPostShuffleInputSize 134217728
spark.sql.adaptive.shuffle.targetPostShuffleRowCount 10000000

结果文件合并

开启合并，将结果文件合并
set spark.sql.mergeFile.enabled=true;
set spark.sql.mergeFile.target.fileSize

orc

spark.hadoop.hive.exec.orc.split.strategy是用来设置spark在读取orc文件时候的策略的，
BI策略以文件为粒度进行split划分；
ETL策略会将文件进行切分，多个stripe组成一个split；
HYBRID策略为：当文件的平均大小大于hadoop最大split值（默认256M）时使用ETL策略，否则使用BI策略。
对于一些较大的ORC表，可能其footer（用于描述整个文件的基本信息、表结构信息、行数、各个字段的统计信息以及各个Stripe的信息）较大，ETL策略可能会导致其从hdfs拉取大量的数据来切分split，甚至会导致driver端OOM，因此这类表的读取建议使用BI策略。
对于一些较小的尤其有数据倾斜的表（这里的数据倾斜指大量stripe存储于少数文件中），建议使用ETL策略

内存

spark.driver.memory/spark.executor.memory
broadcast数据加载在spark.driver.memory内，之后复制到spark.executor.memory内，broadcast数据过大会导致Driver端和Executor端出现内存不足的情况
spark.driver.memoryOverhead/spark.executor.memoryOverhead
container 会预留一部分内存，形式是堆外，用来保证稳定性，主要存储nio buffer，函数栈等一些开销，这部分内存，你不用管堆外还是堆内，开发者用不到，spark也用不到，
所以不用关心，千万不指望调这个参数去提升性能，它的目的是保持运行时的稳定性。
spark.storage.memoryFraction
设置RDD持久化数据在Executor内存中能占的比例，默认Executor60%的内存，可以用来保存持久化的RDD数据。如果内存不够时，可能数据就不会持久化，或者数据会写入磁盘。
参数调优建议：如果Spark作业中，有较多的RDD持久化操作，该参数的值可以适当提高一些，保证持久化的数据能够容纳在内存中。避免内存不够缓存所有的数据，导致数据只能写入磁盘中，降低了性能。但是如果Spark作业中的shuffle类操作比较多，而持久化操作比较少，那么这个参数的值适当降低一些比较合适。此外，如果发现作业由于频繁的gc导致运行缓慢（通过spark web ui可以观察到作业的gc耗时），意味着task执行用户代码的内存不够用，那么同样建议调低这个参数的值。
spark.shuffle.memoryFraction
设置shuffle过程中一个task拉取到上个stage的task的输出后，进行聚合操作时能够使用的Executor内存的比例，默认Executor默认只有20%的内存用来进行该操作。shuffle操作在进行聚合时，如果发现使用的内存超出了这个20%的限制，那么多余的数据就会溢写到磁盘文件中去，此时就会极大地降低性能。
参数调优建议：如果Spark作业中的RDD持久化操作较少，shuffle操作较多时，建议降低持久化操作的内存占比，提高shuffle操作的内存占比比例，避免shuffle过程中数据过多时内存不够用，必须溢写到磁盘上，降低了性能。此外，如果发现作业由于频繁的gc导致运行缓慢，意味着task执行用户代码的内存不够用，那么同样建议调低这个参数的值。
spark.memory.fraction
默认Executor默认只有40%的内存，代表Pool所使用的内存占堆内内存的百分比，这部分内存越大，在大数据量场景下触发的spill次数越少，但同时带来的是非Pool内存的减少，可能会出现OOM。