持续更新
num-executors
该参数用来设置Spark作业总共要用多少Executor进程来执行。Driver在想YARN集群管理器申请资源时,YARN集群管理器会尽可能按照你的设置来在集群的各个工作节点上,启动相应数量的Executor进程。executor-memory
该参数用来设置每个Executor进程的内存。Executor内存的大小,很多时候直接决定了Spark作业的性能,而且跟常见的JVM OOM异常,也有直接的关联。
每个Executor进程的内存设置4G-8G较为合适。但是这只是一个参考值,具体的设置还是得根据不同部门的资源队列来定。可以看看自己团队的资源队列的最大内存限制是多少,num-executors乘以executor-memory,就代表了你的Spark作业申请到的总内存量(也就是所有Executor进程的内存总和),这个量是不能超过队列的最大内存量的。此外,如果你是跟团队里其他人共享这个资源队列,那么申请的总内存量最好不要超过资源队列最大总内存的1/3-1/2,避免你自己的Spark作业占用了队列所有的资源,导致别的同学的作业无法运行。executor-cores
该参数用于设置每个Executor进程的CPU core数量。这个参数决定了每个Executor进程并行执行task线程的能力。因为每个CPU core同一时间只能执行一个task线程,因此每个Executor进程的CPU core数量越多,越能够快速地执行完分配给自己的所有task线程。
Executor的CPU core数量设置为2-4个较为合适。同样得根据不同部门的资源队列来定,可以看看自己的资源队列的最大CPU core限制是多少,再依据设置的Executor数量,来决定每个Executor进程可以分配到几个CPU core。同样建议,如果是跟他人共享这个队列,那么num-executors * executor-cores不要超过队列总CPU core的1/3-1/2左右比较合适,也是避免影响其他同学的作业运行。broadcast join
当大表Join小表时,如果小表足够小,可以将大表分片,分别用小表和大表的分片进行Join,最后汇总,能够大大提升作业性能。
spark.sql.autoBroadcastJoinThreshold 默认值为26214400(25M),如果该表的大小小于该值,则会将小表广播到所有的executor中,使Join快速完成。如果该值设置过大,则会导致executor内存压力过大,容易出现OOM。
ps:ORC格式的表会对数据进行压缩,通常压缩比为2-3左右,但有些表的压缩比会很高,有时可以达到10。需妥善配置该参数,并配合spark.executor.memory,使作业能够顺利执行。spark.dynamicAllocation.enabled:是否开启动态资源分配,默认开启,同时强烈建议用户不要关闭。当开启动态资源分配时,Spark可以根据当前作业的负载动态申请和释放资源。
spark.dynamicAllocation.maxExecutors:开启动态资源分配后,最多可以申请的executor个数。平台默认设置为1000.当在Spark UI中观察到task较多时,可以调大此参数,保证task能够并发执行完成,缩短作业执行时间。
spark.dynamicAllocation.minExecutors:和上面相反,此参数限定了某一时刻executor的最小个数。平台默认是3,即在任何时刻,作业都会保持至少有3个及以上的executor存活,保证任务可以迅速调度。
spark.sql.shuffle.partitions: 在有Join或者聚合等需要shuffle的操作时,从mapper端写出的partition个数,默认设置2000。
如
select a,avg(c) from table group by a
不考虑优化行为,如果一个map端的task中包含有3000个a,根据spark.sql.shuffle.partitions=2000,会将计算结果分成2000份partition,写到磁盘,启动2000个reducer,每个reducer从每个mapper端拉取对应索引的partition。
当作业数据较多时,适当调大该值,当作业数据较少时,适当调小以节省资源。
spark.sql.adaptive.enabled:是否开启调整partition功能,如果开启,spark.sql.shuffle.partitions设置的partition可能会合并到一个reducer中运行。默认开启,可以更好的利用当个executor的性能,还能缓解小文件的问题。
spark.sql.adaptive.shuffle.targetPostShuffleInputSize:和spark.sql.adaptive.enabled配合使用,当开启调整partition功能后,当mapper端两个partition的数据合并后数据量晓雨targetPostShuffleInputSize时,spark会将两个partition进行合并到一个reducer端进行处理。平台默认是64M,用户可根据自身作业的情况调整该值。当调大该值时,一个reducer端task处理的数据量变大,最终产出的数据,存到HDFS上文件也变大;当调小该值时,相反。
spark.sql.adaptive.minNumPostShufflePartitions:当开启调整partition功能后,有时会导致很多分区被合并,为了防止分区过少,可以设置该参数,防止分区过少影响性能。
spark.executor.memory:executor用于缓存数据,代码执行的堆内存及JVM运行时需要的内存。当executor端由于OOM时,多数是spark.executor.memory设置较小引起的。该参数一般可以根据表中单个文件的大小进行估计,但是如果是ORC格式,则需要对文件大小乘以2-3倍。这是由于文件解压后空间要增长。平台默认设置2G。
spark.yarn.executor.memoryOverhead:spark运行还需要一些堆外内存,直接向系统申请,如数据传输时的netty等。Spark根据spark.executor.memory+spark.yarn.executor.memoryOverhead的值向RM申请一个容器,当executor运行时使用的内存超过这个限制时,会被yarn kill掉。在spark UI中相应失败的task错误信息为:
Container killed by YARN for exceeding memory limits. XXX of YYY physical memory used. Consider boosting spark.yarn.executor.memoryOverhead.
此时,适当调大spark.yarn.executor.memoryOverhead。默认设置为1024(1G),该参数单位是Mb。
- spark.sql.windowExce.buffer.spill.threshold:当用户的Sql中包含窗口函数时,并不会把一个窗口中的所有数据全部读进内存,而是维护一个缓存池,当池中的数据条数大于该参数表示的阈值时,spark将数据写到磁盘。该参数如果设置的过小,会导致spark频繁写磁盘,如果设置过大则一个窗口中的数据全都留在内存,有OOM的风险。但是,为了实现快速读入磁盘的数据,spark在每次度磁盘数据时,会保存一个1M的缓存。
ps:当spark.sql.windowExec.buffer.spill.threshold为10时,如果一个窗口有100条数据,则spark会写9次磁盘,在读的时候,会创建9个磁盘reader,每个reader会有一个1M的空间做缓存,也就是说会额外增加9M的空间。
当某个窗口中数据特别多时,会导致写磁盘特别频繁。就会占用很大的内存空间做缓存。因此如果观察到executor的日志中存在大量如下内容,则可以考虑适当调大该参数,平台默认是40960.
pilling data because number of spilledRecords crossed the threshold
- spark.executor.cores:单个executor上可以同时运行的task数据。Spark中的task调度在线程上,该参数决定了一个executor上可以并行执行几个task。这几个task共享一个executor的内存。适当提高该参数的值,可以有效增加程序的并发度,使作业执行的更快,但使executor端的日志变得不易阅读,同时增加executor内存的压力,容易出现OOM。在作业executor端出现OOM时,不过不能增大spark.executor.memory,可以适当降低该值。默认是1.
该参数是executor的并发度,和spark.dynamicAllocation.maxExecutors配合,可以提高整个作业并发度。