SparkConf源码解读
------------恢复内容开始------------
1.主要功能：SparkConf是Spark的配置类，配置spark的application的应用程序，使用(key,value)来进行存储配置信息。
2.主要形式：val conf=new SparkConf()，读取任何spark.*的配置，包括开发人员所设置的配置，因为SparkConf中含有辅助构造器：def this()=this(true)，此辅助构造器中布尔值为true说明读取外部配置信息。在配置单元里可以设置def this()=this(false)，跳过外在配置信息。
3.Spark配置：
Spark中的每一个组件都直接或者间接的使用SparkConf所存储的配置属性，这些属性都存储在数据结构ConcurrentHashMap中即
private val settings=new ConcurrentHashMap[String,String]()
参数配置的优先级为SparkConf > spark-submit和spark-shell命令的参数指定 > 配置文件spark-defaults.conf。
4.如何获取SparkConf配置：
（1）来源于系统参数(即System.getproperties获取的属性)中以Spark.作为前缀的那部分属性
（2）使用SparkConf的API进行设置
（3）从其它SparkConf中克隆
（1）系统属性中配置
在SparkConf中有一个Boolean类型属性loadDefaults，当loadDefaults为true时，将从系统属性中加载Spark配置，代码如下：
/** Create a SparkConf that loads defaults from system properties and the classpath */
def this() = this(true) //构造方法
if (loadDefaults) {
loadFromSystemProperties(false)
}
private[spark] def loadFromSystemProperties(silent: Boolean): SparkConf = {
// Load any spark.* system properties 加载以spark. 开头的系统属性
for ((key, value) <- Utils.getSystemProperties if key.startsWith("spark.")) {
set(key, value, silent)
}
this
}
loadFromSystemProperties
上述代码调用了Utils工具类的getSystemProperties方法，其作用为获取系统的键值对属性，loadFromSystemProperties获取到系统属性后，使用scala守卫过滤出以"spark."字符串为前缀的Key和value并且调用set方法，最终设置到settings中
private[spark] def set(key: String, value: String, silent: Boolean): SparkConf = {
if (key == null) {
throw new NullPointerException("null key")
}
if (value == null) {
throw new NullPointerException("null value for " + key)
}
if (!silent) {
logDeprecationWarning(key)
}
settings.put(key, value)
this
}
配置属性存储到settings中
（2）使用SparkConf配置的API
给SparkConf添加配置的一种常见方式是使用SparkConf提供的API，其中这些API最终实际调用了set的重载方法如：
重载的set方法
/** Set a configuration variable. */
def set(key: String, value: String): SparkConf = {
set(key, value, false)
}
Sparkconf的setMaster，setAppName，setJars，setExecutorEnv，setSparkHome，setAll等方法都是通过上述的set方法完成Spark配置的，如setMaster和setAppName
/**
* The master URL to connect to, such as "local" to run locally with one thread, "local[4]" to
* run locally with 4 cores, or "spark://master:7077" to run on a Spark standalone cluster.
*/
def setMaster(master: String): SparkConf = {
set("spark.master", master)
}
/** Set a name for your application. Shown in the Spark web UI. */
def setAppName(name: String): SparkConf = {
set("spark.app.name", name)
}
添加配置
（3）克隆SparkConf配置
在某些情况下，同一个SparkConf实例中的配置信息需要被多个组件公用，而我们往往会想到的方法是将SparkConf实例定义为全局变量或者通过参数传递给其他组件，但是这样会引入并发问题，虽然settings数据结构为ConcurrentHashMap是线程安全的，而且ConcurrentHashMap也被证明是高并发下性能表现不错的数据结构，但是存在并发，就一定有性能的损失问题，也可以创建一个SparkConf实例b，并将a中的配置信息全部拷贝到b中，这样会浪费内存，导致代码散落在程序的各个部分。
SparkConf继承了Cloneable物质并实现了clone方法，可以通过Cloneable物质提高代码的可利用性
Cloneable物质
class SparkConf(loadDefaults: Boolean) extends Cloneable with Logging {
def this() = this(true)
/\*\* Copy this object \*/
override def clone: SparkConf = {
val cloned = new SparkConf(false)
settings.entrySet().asScala.foreach { e =>
cloned.set(e.getKey(), e.getValue(), true)
}
cloned
} }
5.SparkConf中的优化措施：
（1）SparkConf在spark1.3版本时，spark.yarn.applicationMaster.waitTries是SparkContext初始化失败最大次数，现如今配置成spark.yarn.am.waitTime，重试等待时间，默认为10s，依次递增为20s，30s
waitTries变成waitTime
"spark.yarn.am.waitTime" -> Seq(
AlternateConfig("spark.yarn.applicationMaster.waitTries", "1.3",
// Translate old value to a duration, with 10s wait time per try.
translation = s => s"${s.toLong * 10}s"))
（2）spark.reducer.maxSizeInFlight，默认为48M，此参数用于设置shuffle过程中每个reduce任务需要一个缓冲区来接受map任务输出的结果，而这个缓冲区决定了每次能够拉取多少数据。调优建议：如果作业可用的内存资源较为充足的情况下，可以适当增加这个参数的大小，如设置成96M，从而减少拉取数据的次数，也可以减少网络传输的次数。
"spark.reducer.maxSizeInFlight" -> Seq(
AlternateConfig("spark.reducer.maxMbInFlight", "1.4"))
spark.reducer.maxSizeInFlight
（3）spark.kryoserializer.buffer
在Spark的架构中，在网络中传递的或者缓存在内存、硬盘中的对象需要进行序列化操作，序列化的作用主要是利用时间换空间，主要应用于：分发给Executor上的Task，需要缓存的RDD(前提是使用序列化方式缓存的)，广播变量，Shuffle过程中的数据缓存，算子函数中使用到的外部变量。大大减少了数据在内存、硬盘中占用的空间，减少了网络数据传输的开销，并且可以精确的推测内存使用情况，降低GC频率。但是把数据序列化为字节数组、把字节数组反序列化为对象的操作，是会消耗CPU、延长作业时间的，从而降低了Spark的性能。使用kryo序列化三步走策略：
第一步：设置序列化使用的库：conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer"); //使用Kryo序列化库
第二步：在该库中注册用户定义的类型：conf.set("spark.kryo.registrator", toKryoRegistrator.class.getName()); //在Kryo序列化库中注册自定义的类集合
第三步：在自定义类中实现KryoRegistrator接口的registerClasses方法
public static class toKryoRegistrator implements KryoRegistrator {
public void registerClasses(Kryo kryo) {
kryo.register(tmp1.class, new FieldSerializer(kryo, tmp1.class)); //在Kryo序列化库中注册自定义的类
kryo.register(tmp2.class, new FieldSerializer(kryo, tmp2.class)); //在Kryo序列化库中注册自定义的类
}
}
（4）spark.shuffle.file.buffer和spark.shuffle.consolidateFiles
spark中shuffle输出的ShuffleMapTask为每个ResultTask创建对应的Bucket，ShuffleMapTask产生的结果会根据设置partitioner的到对应的BucketId，然后填充到相应的Bucket中。每个ShuffleMapTask的输出结果可能包含所有ResultTask需要的数据，所以每个ShuffleMapTask创建的Bucket的数目和ResultTask的数目是相当的。
ShuffleMapTask创建的Bucket对应磁盘上的一个文件，用于存储结果，此文件被称为BlockFile。通过 spark.shuffle.file.buffer.kb属性配置的缓冲区就是用来创建BufferedOutputStream输出流的，如果配置文件中设置了spark.shuffle.consolidateFiles属性为true，则ShuffleMapTask所产生的Bucket就不一定单独对应一个文件了，而是对应文件的一部分，这样做会大量减少产生的BlockFile文件数量。
------------恢复内容结束------------
1.主要功能：SparkConf是Spark的配置类，配置spark的application的应用程序，使用(key,value)来进行存储配置信息。
2.主要形式：val conf=new SparkConf()，读取任何spark.*的配置，包括开发人员所设置的配置，因为SparkConf中含有辅助构造器：def this()=this(true)，此辅助构造器中布尔值为true说明读取外部配置信息。在配置单元里可以设置def this()=this(false)，跳过外在配置信息。
3.Spark配置：
Spark中的每一个组件都直接或者间接的使用SparkConf所存储的配置属性，这些属性都存储在数据结构ConcurrentHashMap中即
private val settings=new ConcurrentHashMap[String,String]()
4.如何获取SparkConf配置：
（1）来源于系统参数(即System.getproperties获取的属性)中以Spark.作为前缀的那部分属性
（2）使用SparkConf的API进行设置
（3）从其它SparkConf中克隆
（1）系统属性中配置
在SparkConf中有一个Boolean类型属性loadDefaults，当loadDefaults为true时，将从系统属性中加载Spark配置，代码如下：
/** Create a SparkConf that loads defaults from system properties and the classpath */
def this() = this(true) //构造方法
if (loadDefaults) {
loadFromSystemProperties(false)
}
private[spark] def loadFromSystemProperties(silent: Boolean): SparkConf = {
// Load any spark.* system properties 加载以spark. 开头的系统属性
for ((key, value) <- Utils.getSystemProperties if key.startsWith("spark.")) {
set(key, value, silent)
}
this
}
loadFromSystemProperties
上述代码调用了Utils工具类的getSystemProperties方法，其作用为获取系统的键值对属性，loadFromSystemProperties获取到系统属性后，使用scala守卫过滤出以"spark."字符串为前缀的Key和value并且调用set方法，最终设置到settings中
/** Create a SparkConf that loads defaults from system properties and the classpath */
def this() = this(true) //构造方法
if (loadDefaults) {
loadFromSystemProperties(false)
}
private[spark] def loadFromSystemProperties(silent: Boolean): SparkConf = {
// Load any spark.* system properties 加载以spark. 开头的系统属性
for ((key, value) <- Utils.getSystemProperties if key.startsWith("spark.")) {
set(key, value, silent)
}
this
}
（2）使用SparkConf配置的API
给SparkConf添加配置的一种常见方式是使用SparkConf提供的API，其中这些API最终实际调用了set的重载方法如：
重载的set方法
/** Set a configuration variable. */
def set(key: String, value: String): SparkConf = {
set(key, value, false)
}
Sparkconf的setMaster，setAppName，setJars，setExecutorEnv，setSparkHome，setAll等方法都是通过上述的set方法完成Spark配置的，如setMaster和setAppName
/**
* The master URL to connect to, such as "local" to run locally with one thread, "local[4]" to
* run locally with 4 cores, or "spark://master:7077" to run on a Spark standalone cluster.
*/
def setMaster(master: String): SparkConf = {
set("spark.master", master)
}
/** Set a name for your application. Shown in the Spark web UI. */
def setAppName(name: String): SparkConf = {
set("spark.app.name", name)
}
添加配置
（3）克隆SparkConf配置
在某些情况下，同一个SparkConf实例中的配置信息需要被多个组件公用，而我们往往会想到的方法是将SparkConf实例定义为全局变量或者通过参数传递给其他组件，但是这样会引入并发问题，虽然settings数据结构为ConcurrentHashMap是线程安全的，而且ConcurrentHashMap也被证明是高并发下性能表现不错的数据结构，但是存在并发，就一定有性能的损失问题，也可以创建一个SparkConf实例b，并将a中的配置信息全部拷贝到b中，这样会浪费内存，导致代码散落在程序的各个部分。
SparkConf继承了Cloneable物质并实现了clone方法，可以通过Cloneable物质提高代码的可利用性
Cloneable物质
class SparkConf(loadDefaults: Boolean) extends Cloneable with Logging {
def this() = this(true)
/\*\* Copy this object \*/
override def clone: SparkConf = {
val cloned = new SparkConf(false)
settings.entrySet().asScala.foreach { e =>
cloned.set(e.getKey(), e.getValue(), true)
}
cloned
} }
5.SparkConf中的优化措施：
（1）SparkConf在spark1.3版本时，spark.yarn.applicationMaster.waitTries是SparkContext初始化失败最大次数，现如今配置成spark.yarn.am.waitTime，重试等待时间，默认为10s，依次递增为20s，30s
waitTries变成waitTime
"spark.yarn.am.waitTime" -> Seq(
AlternateConfig("spark.yarn.applicationMaster.waitTries", "1.3",
// Translate old value to a duration, with 10s wait time per try.
translation = s => s"${s.toLong * 10}s"))
（2）spark.reducer.maxSizeInFlight，默认为48M，此参数用于设置shuffle过程中每个reduce任务需要一个缓冲区来接受map任务输出的结果，而这个缓冲区决定了每次能够拉取多少数据。调优建议：如果作业可用的内存资源较为充足的情况下，可以适当增加这个参数的大小，如设置成96M，从而减少拉取数据的次数，也可以减少网络传输的次数。
spark.reducer.maxSizeInFlight
（3）spark.kryoserializer.buffer
在Spark的架构中，在网络中传递的或者缓存在内存、硬盘中的对象需要进行序列化操作，序列化的作用主要是利用时间换空间，主要应用于：分发给Executor上的Task，需要缓存的RDD(前提是使用序列化方式缓存的)，广播变量，Shuffle过程中的数据缓存，算子函数中使用到的外部变量。大大减少了数据在内存、硬盘中占用的空间，减少了网络数据传输的开销，并且可以精确的推测内存使用情况，降低GC频率。但是把数据序列化为字节数组、把字节数组反序列化为对象的操作，是会消耗CPU、延长作业时间的，从而降低了Spark的性能。使用kryo序列化三步走策略：
第一步：设置序列化使用的库：conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer"); //使用Kryo序列化库
第二步：在该库中注册用户定义的类型：conf.set("spark.kryo.registrator", toKryoRegistrator.class.getName()); //在Kryo序列化库中注册自定义的类集合
第三步：在自定义类中实现KryoRegistrator接口的registerClasses方法
public static class toKryoRegistrator implements KryoRegistrator {
public void registerClasses(Kryo kryo) {
kryo.register(tmp1.class, new FieldSerializer(kryo, tmp1.class)); //在Kryo序列化库中注册自定义的类
kryo.register(tmp2.class, new FieldSerializer(kryo, tmp2.class)); //在Kryo序列化库中注册自定义的类
}
}
（4）spark.shuffle.file.buffer和spark.shuffle.consolidateFiles
spark中shuffle输出的ShuffleMapTask为每个ResultTask创建对应的Bucket，ShuffleMapTask产生的结果会根据设置partitioner的到对应的BucketId，然后填充到相应的Bucket中。每个ShuffleMapTask的输出结果可能包含所有ResultTask需要的数据，所以每个ShuffleMapTask创建的Bucket的数目和ResultTask的数目是相当的。
ShuffleMapTask创建的Bucket对应磁盘上的一个文件，用于存储结果，此文件被称为BlockFile。通过 spark.shuffle.file.buffer.kb属性配置的缓冲区就是用来创建BufferedOutputStream输出流的，如果配置文件中设置了spark.shuffle.consolidateFiles属性为true，则ShuffleMapTask所产生的Bucket就不一定单独对应一个文件了，而是对应文件的一部分，这样做会大量减少产生的BlockFile文件数量。
手机扫一扫
移动阅读更方便
你可能感兴趣的文章