Spark 配置整理

Spark 的配置有很多，这里一方面总结一下官方文档中的内容，一方面将网上查到的资料中用到的针对特定问题的配置整理一下。

先看一下官网的配置：http://spark.apache.org/docs/latest/configuration.html

spark 配置可分为三层： spark properties、environment variables、还有logging

 

spark properties 是由用户自己设置的，在任务中通过 SparkConf 类设置：

val conf = new SparkConf()
             .setMaster("local")
             .setAppName("CountingSheep")
             .set("spark.executor.memory", "1g")val sc = new SparkContext(conf)

或者 你可以通过提交命令设置，这个时候 SparkConf 对象就不设置相关配置：

./bin/spark-submit --name "My app" --master local[4] --conf spark.shuffle.spill=false
  --conf "spark.executor.extraJavaOptions=-XX:+PrintGCDetails -XX:+PrintGCTimeStamps" myApp.jar

在提交作业的时候能设置哪些配置呢？可以通过 bin/spark-submit --help 查看

spark-submit 还会加载 conf/spark-defaults.conf 中的配置

spark.master            spark://5.6.7.8:7077
spark.executor.memory   512m
spark.eventLog.enabled  true
spark.serializer        org.apache.spark.serializer.KryoSerializer

如何查看已经配置的配置项？
http://Spark-master:ui-port/history/application-id/environment/
需要关注的一些设置：


spark.executor.memory  默认 512M，设置executor 的 jvm的内存大小


spark.executor.extraJavaOptions 额外的jvm设置，比如说gc策略，但是这里不能设置 jvm 的内存大小


spark.executor.extraLibraryPath  设置额外的executor运行lib目录


spark.shuffle.consolidateFiles 默认 false ， 如果设置成true，能够整合shuffle阶段产生的中间文件，当reduce任务比较多的时候能提高性能


spark.shuffle.spill 默认true，与下面的配置配合使用，用来限制shuffle阶段，reduce占用内存的大小，如果超过内存限制就将数据刷到磁盘


spark.shuffle.memoryFraction 默认 0.2，如果上面的配置设置为true，shuffle阶段的内存占用总的内存的比例，如果超过这个比例，就将数据存到磁盘上。如果经常将数据刷到磁盘，可以将这个比例提高。


spark.shuffle.spill.compress 默认为true，shuffle阶段将中间结果刷入磁盘是否压缩


spark.shuffle.compress 默认为true，shuffle阶段中间结果是否压缩，这两种压缩都是使用的：spark.io.compression.codec 压缩算法


spark.shuffle.file.buffer.kb 默认32，每一个在内存中的shuffle file的大小，不把中间文件都刷到磁盘，可以减少对磁盘的seek和系统调用


spark.reducer.maxMbInFlight 默认48，对于reduce端，默认的会有5个线程负责拉取shuffle的中间结果，这5个线程分享这48M的内存的缓冲空间。如果内存不太大，最好设置的小一点。


spark.shuffle.manager 默认HASH，表示使用hash-based shuffle 算法，对于spark1.1版本，可以使用SORT，sort-based shuffle


spark.eventLog.enabled 默认false，记录spark events，当application运行结束，可用于重建 web ui


spark.eventLog.dir 默认 /tmp/spark-events， 如果spark.eventLog.enabled 为true，用来存放event log的地址，可以设置成hdfs的目录


spark.io.compression.codec 默认 snappy，spark用到的压缩地方统一的压缩算法，三种可选配置：org.apache.spark.io.LZ4CompressionCodec,org.apache.spark.io.LZFCompressionCodec,
org.apache.spark.io.SnappyCompressionCodec


spark.default.parallelism 不同的模式使用的默认值不同，一般情况默认为2，使用到shuffle的任务，比如groupByKey、reduceByKey，如果用户没有设置并行度，则使用这个配置的默认并行度
spark.akka.frameSize 默认10，单位兆，akka传递消息的最大值，如果你的任务需要将比较大的结果传给driver，可以提高这个值，比如说在比较大的数据集上调用.collect()


spark.akka.threads 默认4，用于akka通信的线程数，如果集群比较大，机器内核比较多，可以增大这个值


spark.task.cpus 默认1，每个task分配的cpu


spark.task.maxFailures 默认4，task重试的次数


spark.scheduler.mode 默认FIFO，同一个SparkContext提交的jobs之间使用的调度策略，可以设置成FAIR


spark.locality.wait 默认3000，单位毫秒，为了起一个data local 的任务可以等待的时间。


spark.scheduler.revive.interval 默认 1000，单位毫秒，调度器激活task运行的时间间隔


对于在YARN集群上运行spark任务，还有一些特殊的配置需要注意：


spark.yarn.submit.file.replication 默认3，提交到yarn上的任务所需要的jar或者文件的副本数


spark.yarn.preserve.staging.files 默认false，如果设成true，将在任务运行完保存spark相关文件，比如 spark jar，app jar


spark.yarn.executor.memoryOverhead 默认384，单位兆，executor 分配的jvm内存，显然有点小


spark.yarn.driver.memoryOverhead 默认384，单位兆，driver的内存，也有点小


spark.yarn.jar  放spark jar文件的路径，当提交spark任务到yarn的时候，会将spark的jar和app的jar拷贝到hdfs的app的目录中，但是每一个任务都会用到spark的jar或者一些共用的jar包，所以，把这些共用jar包放到一个公共的位置，这样就省的每个任务都提交一次。例如：hdfs://some/path