优化之前必须对要处理的数据量有一个充分的认识，比如：
1.哪些数据是需要的，哪些是不需要的，数据过滤可否提前；
2.多表链接的时候，先连接小表达到过滤较多数据的目的；
3.数据格式是什么样的；
4.数据是如何分布的，有多少分区，分区之间是否平衡；
5.执行计划是什么样的（通过Spark UI、Spark history Server、explain、toDebugString等）。
只有充分了解要处理的数据才能够做好优化。
1.RDD重新分区
针对大量小分区的RDD，使用RDD重分区函数coalesce将小分区合并成大分区；同样当分区数据量过大也可以使用重新分区，增加分区数量，提高并行计算能力。一般推荐每个分区12MB左右的数据，默认值为128MB。
具体措施：
1.1 读取数据时：
a.读取文件的时候设置分区（sparkContext.textFile("filePath",200)）；
b.使用Catalyst优化器管理DataFrame、DataSet的基础RDD的分区.
1.2 Transformation算子创建RDD的时：
Transformation算子创建的子RDD的分区数量依赖于父RDD.
a.通过修改spark.default.parappelism来制定默认值；
b.如果Transformation算子支持设置numPartitions，可以设置numPartitions，如reduceByKey();
备注：
coalesce默认不执行shuffle，因此只适用于分区个数由大变小的场景；
repartition也是调用了coalesce方法，只是shuffle为true，因此repartition执行了shuffle，适用于分区个数由小变大的场景。
2.并行度
2.1 通过配置和代码来设置task数量。task数量太多导致task的启动和切换开销；task数量太小，无法利用并行计算的能力；
2.2 使用hive sql时可指定并行度。
3.DAG优化
a.一个Stage尽量容纳更多的算子，减少Shuffle的发生；
b.将经常用到的数据缓存到内存，然后复用已经cache的数据。
4.倾斜问题
倾斜问题是指在个别分区上，执行时间过长，最长时长远大于平均时长。
数据倾斜（比如分区key或者分区函数设计的不好），task倾斜（数据倾斜导致任务倾斜），task执行速度倾斜(解决方式：去掉执行过长的task任务节点，重新分配调度)。
解决方式：
4.1.增大task数量，减少每个分区的数据量；
4.2.对特殊的key进行处理，如空值映射为特定key，分发到不同节点进行特殊处理；
4.3.将数据量大的表划分成多个小表，分成多个阶段进行；
4.4.拆分RDD:将倾斜数据与元数据分离，分成2个job进行计算。
4.5.优化聚集操作，避免使用shuffle。
4.6.使key随机化，如每个key后面加一个0-N的随机数，这样讲task拆分；
4.7.对4.6进行优化，只对数据量比较大的key进行key随机化操作，需要统计key的数据量，避免数据量小的没必要拆分key也没拆分；
4.8 针对一个RDD数据量小的情况，使用map join替换reduce join(不适合两个RDD都很大的场景);
5.JVM调优
5.1 设计好的数据结构，避免使用链式集合；
5.2 减少对象嵌套；
5.3 尽量使用数字ID或者枚举对象作为key，尽量避免使用字符串；
5.4 JVM的GC调优；
5.5 磁盘临时空间优化，当执行Shuffle或者内存不够缓存RDD时，RDD的数据会写到磁盘临时目录中；
5.5 JVM重复使用，因为一个task默认启动一个JVM，task结束之后，销毁JVM，JVM重复使用避免JVM的启动和销毁对资源的消耗。
6.网络传输优化
6.1 增大task的分发缓存大小，避免过大的task造成缓存区溢出；
6.2 小数据量表数据可以直接广播到各个Exector节点上，这样该Executor节点上的N个task共享广播数据，避免每个task都从Driver端获取该数据；
6.3 广播变量的经过是，将数据读取到driver端（类似于Collect函数），在分发到Executor上，不合理的广播变量会导致driver端的OOM。
6.4 Collect结果过大时要使用序列化，因为collect函数将每个分区变为数组，返回主节点，并将所有分区的数组合并成一个分组。
7.序列化和压缩
7.1 序列化存储RDD，官方推荐kyro序列化方式；
7.2 压缩存储RDD，spark支持LZF和snappy两种压缩方式；
7.2 Hadoop权衡行式存储和列式存储，行式存储查询过滤需要将所有数据遍历，列式存储在查询时，只需要读取需要的列块，因此查询性能更好，而且具有更好的压缩效率。
不同应用场景选择合适的存储格式:
8.方法的优化
8.1 尽量避免使用reduce方法，而使用reduceByKey
因为reduce是Action算子，是聚集函数，会导致结果汇聚到主节点；而reduceByKey是transformation算子，使计算变成分布式计算。
8.2 针对shuffle操作（如sortByKey，groupByKey，reduceByKey， join等），可以通过增加并行度
这样每个task处理的数据量就会相对减少，占用内存就会减少，避免OutOfMemory产生。
8.3 join和map join的选择
map join通过Local Task将小表读入内存，生成HashTableFiles，并压缩上传至Distributed Cache中（map join 不适合两个都是大表的情况，否则会导致内存溢出）。
在Map阶段，每个Mapper从Distributed Cache读取HashTableFiles到内存中，顺序扫描大表，在Map阶段实现join，将数据传递给下一个MapReduce任务，避免了shuffle，加快处理效率。
8.4 cache和persist选择
cache其实只是调用了persist()方法，默认缓存到内存中（persist(StorageLevel.MEMORY_ONLY)）；
persist支持多种缓存方式，具体如下：
object StorageLevel {
val NONE = new StorageLevel(false, false, false, false)
val DISK_ONLY = new StorageLevel(true, false, false, false)
val DISK_ONLY_2 = new StorageLevel(true, false, false, false, 2)
val MEMORY_ONLY = new StorageLevel(false, true, false, true)
val MEMORY_ONLY_2 = new StorageLevel(false, true, false, true, 2)
val MEMORY_ONLY_SER = new StorageLevel(false, true, false, false)
val MEMORY_ONLY_SER_2 = new StorageLevel(false, true, false, false, 2)
val MEMORY_AND_DISK = new StorageLevel(true, true, false, true)
val MEMORY_AND_DISK_2 = new StorageLevel(true, true, false, true, 2)
val MEMORY_AND_DISK_SER = new StorageLevel(true, true, false, false)
val MEMORY_AND_DISK_SER_2 = new StorageLevel(true, true, false, false, 2)
val OFF_HEAP = new StorageLevel(true, true, true, false, 1)
...
}
class StorageLevel private (private var _useDisk: Boolean,private var_useMemory: Boolean, private var _useOffHeap: Boolean, private var _deserialized: Boolean, private var _replication: Int = 1) extends Externalizable {
...
def useDisk: Boolean = _useDisk
def useMemory: Boolean = _useMemory
def useOffHeap: Boolean = _useOffHeap
def deserialized: Boolean = _deserialized
def replication: Int = _replication
...
}
备注：
a._2，代表的是将每个持久化的数据，都复制一份副本，并将副本保存到其他节点上，实现容错，避免因为一个RDD分区意外丢失导致所有数据重新计算;
b._SER 代表RDD的每个partition会被序列化成一个字节数组，节省内存。
9.文件大小
9.1 对于大文件，进行拆分，提高并行度
9.2 对于小文件进行合并
wholeTextFile方式将小文件都读入到一个文件中。
10.Catalyst优化器：
10.1 Catalyst优化器的目标：
a.减少executor之间的数据传输；
b.减少shuffle操作;
c.将尽可能多的操作纳入到同一个stage中；
d.运行时为整个stage生成代码。
10.2 Catalyst优化器优化动作：
a.filter前置；
b.select前置；
c.减少shuffle数据量，如聚合之前先内部聚合，join的时候选择最优方式和顺序；
d.常量折叠：编译的时候进行常量计算，将其折叠为文本；
e.其他：如Decimal计算转为long计算。
备注：Lambda影响Catalyst优化器优化
11.广播（broadcast）
广播（broadcast）可以将比较小的RDD或者DataFrame广播到各个机器节点上，在执行join的过程中可以避免shuffle过程，从而降低通信成本，提高执行效率。
在SparkSQL中，可以通过设置spark.sql.autoBroadcastJoinThreshold（默认值10MB），这样当DataFrame对应表的统计信息（有可能不是实际的表存储容量）小于spark.sql.autoBroadcastJoinThreshold，SparkSQL将join转为broadcast Join。可以适当提高spark.sql.autoBroadcastJoinThreshold的值，来提高广播的效率，但是由于广播数据需要经过Driver端，会给Driver造成存储压力，有可能会导致driver端的存储溢出，一般需要提高spark.driver.memory的值。
另外，当表的统计信息更新不及时的时候（如表的分区正在写入数据，spark判断是否需要广播该表，读取统计信息的时候，统计信息中的数据量还小于spark.sql.autoBroadcastJoinThreshold的值，当执行SparkSQL的时候，该表的数据量已经很大，SparkSQL仍然会broadcast该表），有可能会出现广播大表的风险，这种情况一般会出现driver端内存溢出。此时应当配置spark.sql.statistics.fallBackToHdfs=true，即Spark在执行的时候，直接读取HDFS文件的大小，而不是读取表的描述统计信息来进行广播判断的依据（曾经遇到过这样的线上案例，惨痛的教训）。
备注：可以通过hint设置，强制将指定表广播出去，不考虑spark.sql.autoBroadcastJoinThreshold的值。
12.资源分配：
a.yarn相关：
yarn.nodemanager.resource.memory-mb:控制每个节点可用内存大小；
yarn.nodemanager.resource.cpu-vcores:控制每个节点可用最大CPU数量；
b.Spark相关：
num-executors:executors的数量，可以设置Dynamic Allocation动态分配；
spark.dynamicAllocation.enabled // 是否启动Dynamic Allocation
spark.dynamicAllocation.minExecutors// 最小executor数量
spark.dynamicAllocation.maxExecutors// 最大executor数量
spark.dynamicAllocation.initialExecutors// 初始executor数量
executors-cores:每个executor的CPU vcore的数量，一般推荐<=5;
executors-memory:每个executor的内存大小，推荐64GB是上线，一般一个cpu配置4GB内存；
可以通过spark参数配置各区域大小，通过TaskMemoryManager查看各个部分使用量。
13.driver端：
1.有些数据是需要driver端传输数据的，如broadcast、shuffle时，数据是由driver端转发的，可以增加spark.driver.memory减少driver端的存储压力；
2.避免数据汇集到driver端的算子，如collect()(只适合少量数据)、toPandas()、reduce()（用treeReduce替换，可以在executor中先进行计算）、accumulators()。
