Spark配置

Spark提供三个位置配置系统:

1. Spark属性控制大多数应用程序参数，可以通过使用SparkConf对象或通过Java系统属性进行设置。
2. 可以使用环境变量在每个节点上通过conf/spark-env.sh脚本设置每台机器的设置，例如IP地址。
3. 可以通过log4jb .properties配置日志记录。

Spark属性控制大多数应用程序设置，并为每个应用程序单独配置。这些属性可以直接设置SparkConf传递给你的SparkContext。SparkConf允许您配置一些常用属性(例如主URL和应用程序名称)，以及通过set()方法配置任意键值对。例如，我们可以用如下两个线程初始化一个应用程序:

val conf = new SparkConf()
             .setMaster("local[2]")
             .setAppName("CountingSheep")
val sc = new SparkContext(conf)

注意，我们使用本地[2]运行，这意味着两个线程,表示“最小”并行性。
单位格式：

时间：
25ms (milliseconds)
5s (seconds)
10m or 10min (minutes)
3h (hours)
5d (days)
1y (years)
字节大小：
1b (bytes)
1k or 1kb (kibibytes = 1024 bytes)
1m or 1mb (mebibytes = 1024 kibibytes)
1g or 1gb (gibibytes = 1024 mebibytes)
1t or 1tb (tebibytes = 1024 gibibytes)
1p or 1pb (pebibytes = 1024 tebibytes)

动态加载Spark属性

在某些情况下，您可能希望避免在SparkConf中硬编码某些配置。例如，如果您想运行相同的应用程序与不同的主控或不同的内存数量。Spark允许你简单地创建一个空的conf:

val sc = new SparkContext(new SparkConf())

然后，你可以在运行时提供配置值:

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

Spark shell和Spark-submit工具支持两种动态加载配置的方式。第一个是命令行选项，如——master，如上所示。Spark-submit可以接受使用——conf/-c标志的任何Spark属性，但对在启动Spark应用程序时起作用的属性使用特殊标志。运行/bin/spark-submit ——help将显示这些选项的完整列表。

.bin/spark-submit还将从conf/spark-defaults.conf中读取配置选项，其中每一行由一个键和一个由空格分隔的值组成。例如:

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

任何指定为标志或属性文件中的值都将传递给应用程序，并与通过SparkConf指定的值合并。直接在SparkConf上设置的属性优先级最高，然后是传递给spark-submit或spark-shell的标志，然后是spark-defaults.conf文件中的选项。一些配置键从早期版本的Spark开始重新命名;在这种情况下，旧的键名仍然被接受，但其优先级低于新键的任何实例。即:

（程序中配置）>（启动时配置）>（配置文件配置）（新>旧）

Spark属性主要可以分为两种:一种是与部署相关的，比如“Spark.driver.memory”、“spark.executor.instances”，这类属性在运行时通过SparkConf编程设置时可能不受影响，或者行为取决于您选择的集群管理器和部署模式，因此建议通过配置文件或spark-submit命令行选项设置;另一个主要与Spark运行时控制相关，如“spark.task.maxFailures”。这类属性可以用两种方式设置。
详细配置情况，可查询官网文档：https://spark.apache.org/docs/latest/configuration.html

可用配置

Spark SQL

运行时

运行时SQL配置是每会话的、可变的Spark SQL配置。它们可以通过配置文件和带有——conf/-c前缀的命令行选项设置初始值，或者通过设置用于创建SparkSession的SparkConf。此外，它们可以通过set命令设置和查询，并通过RESET命令或运行时SparkSession.conf的setter和getter方法保持初始值。

val sparkConf = new SparkConf()
                    .setMaster("local[*]")
                    .setAppName("sparkSQL")
val spark = SparkSession.builder()
                  .config(sparkConf)
                  .getOrCreate()

静态

静态SQL配置是跨会话的、不可变的Spark SQL配置。它们可以通过配置文件和带有——conf/-c前缀的命令行选项设置最终值，或者通过设置用于创建SparkSession的SparkConf。外部用户可以通过SparkSession.conf或set命令查询静态sql配置值，例如set spark.sql.extensions;，但不能设置/取消设置它们。

Spark Streaming

    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("sparkStreaming")
    // 创建时需要传递两个参数
    // 1.环境配置
    // 2.批处理周期(采集周期)
    val ssc = new StreamingContext(sparkConf, Seconds(3))

集群管理

Spark中的每个集群管理器都有额外的配置选项。配置可以在每个模式的页面上找到:

YARN
Mesos
Kubernetes
Standalone Mode

环境变量

Spark的某些设置可以通过环境变量进行配置，环境变量从Spark安装目录下的conf/Spark-env.sh脚本(或conf/Spark-env.sh)中读取。cmd窗口)。在单机模式和Mesos模式下，该文件可以提供特定于计算机的信息，如主机名。当运行本地Spark应用程序或提交脚本时，它也是来源。
注意:安装Spark时，conf/Spark-env.sh默认不存在。但可以拷贝conf/spark-env.sh.template来创建它。确保你的拷贝是可执行的。

注意:在集群模式下在YARN上运行Spark时，需要使用Spark.YARN.appMasterEnv设置环境变量。在你的conf/spark-defaults.conf文件中的[EnvironmentVariableName]属性。在spark-env.sh中设置的环境变量在集群模式下不会反映在YARN Application Master进程中。更多信息请参见与yarn相关的Spark属性。

配置日志

Spark使用log4j进行日志记录。您可以通过添加log4j2来配置它。conf目录下的属性文件。一种开始的方法是复制现有的log4j2.properties.template。

重写配置目录

如果需要指定默认的“SPARK_HOME/conf”目录以外的其他配置目录，可以通过设置“SPARK_CONF_DIR”。Spark将使用配置文件(spark-defaults.conf, spark-env.sh, log4j2.properties等)。

继承Hadoop集群配置

如果你计划使用Spark从HDFS进行读写操作，有两个Hadoop配置文件应该包含在Spark的类路径中:

hdfs-site.xml,
core-site.xml,

这些配置文件的位置在不同的Hadoop版本中是不同的，但是一个常见的位置是在/etc/Hadoop/conf中。有些工具动态地创建配置，但提供了下载配置副本的机制。
要使这些文件对Spark可见，请将$SPARK_HOME/conf/Spark-env.sh中的HADOOP_CONF_DIR设置到包含配置文件的位置。

hadoop/hive配置

如果您的Spark应用程序正在与Hadoop、Hive或两者交互，那么在Spark的类路径中可能存在Hadoop/Hive配置文件。

多个运行的应用程序可能需要不同的Hadoop/Hive客户端配置。您可以在Spark的类路径中为每个应用程序复制和修改hdfs-site.xml, core-site.xml, yarn-site.xml, hive-site.xml。在运行在YARN上的Spark集群中，这些配置文件是在集群范围内设置的，应用程序无法安全地更改它们。

更好的选择是以spark.hadoop.的形式使用spark hadoop属性。并以spark.hive.形式的spark hive属性。例如，增加配置“spark.hadoop.abc.def=xyz”表示增加hadoop属性“abc.def=xyz”，增加配置“spark.hive.abc=xyz表示添加hive属性“hive.abc =xyz”。它们可以被认为与普通的spark属性相同，可以在$SPARK_HOME/conf/spark-defaults.conf中设置

在某些情况下，您可能希望避免在SparkConf中硬编码某些配置。例如，Spark允许您简单地创建一个空的conf并设置Spark/Spark hadoop/ Spark hive属性。

val conf = new SparkConf().set("spark.hadoop.abc.def", "xyz")
val sc = new SparkContext(conf)

此外，您可以在运行时修改或添加配置:

./bin/spark-submit  
  --name "My app"  
  --master local[4]   
  --conf spark.eventLog.enabled=false  
  --conf "spark.executor.extraJavaOptions=-XX:+PrintGCDetails -XX:+PrintGCTimeStamps"  
  --conf spark.hadoop.abc.def=xyz 
  --conf spark.hive.abc=xyz
  myApp.jar

自定义资源调度与配置概述

gpu和其他加速器已广泛用于加速特殊工作负载，如深度学习和信号处理。Spark现在支持请求和调度通用资源，如gpu，但有一些注意事项。当前的实现要求资源具有可由调度器分配的地址。它需要您的集群管理器支持并正确配置资源。

有一些配置可以为驱动程序请求资源:spark.driver.resource.{resourceName}。spark.executor.resource.{resourceName}为执行器请求资源。spark.task.resource.{resourceName}.amount为每个任务指定需求。spark.driver.resource.{resourceName}在YARN、Kubernetes和Spark Standalone上的客户端驱动程序上需要discoveryScript配置。spark.executor.resource。{resourceName}。discoveryScript配置对于YARN和Kubernetes是必需的。Kubernetes还需要spark.driver.resource.{resourceName}.vendor/spark.executor.resource。{resourceName}.vendor。请参阅上面的配置描述以获得关于每种配置的更多信息。

Spark将使用指定的配置来首先从集群管理器请求具有相应资源的容器。一旦它获得了容器，Spark在该容器中启动一个Executor，它将发现容器中有什么资源以及与每个资源关联的地址。执行程序将向驱动程序注册并向该执行程序报告可用的资源。然后，Spark调度器可以将任务调度到每个Executor，并根据用户指定的资源需求分配特定的资源地址。用户可以使用TaskContext.get().resources api查看分配给任务的资源。在驱动程序上，用户可以看到SparkContext资源调用所分配的资源。然后由用户决定使用分配的地址来执行他们想要的处理，或者将这些地址传递到他们正在使用的ML/AI框架中。

查看您的集群管理器特定页面了解每种模式的要求和详细信息——YARN、Kubernetes和独立模式。它目前不能用于Mesos或本地模式。还请注意，不支持具有多个工作者的本地集群模式(请参阅独立文档)。

stage级别调度概述

阶段级调度特性允许用户在阶段级指定任务和执行程序资源需求。这允许使用具有不同资源的执行程序运行不同的阶段。一个主要的例子是ETL阶段使用只有cpu的执行程序运行，下一个阶段是需要gpu的ML阶段。阶段级调度允许用户在ML阶段运行时请求不同的具有gpu的执行程序，而不必在应用程序开始时获取具有gpu的执行程序，并且这些执行程序在ETL阶段运行时处于空闲状态。这只适用于Scala、Java和Python中的RDD API。当启用动态分配时，它在YARN和Kubernetes上可用。有关更多实现细节，请参见YARN页面或Kubernetes页面。

看RDD,withResources和ResourceProfileBuilder API用于使用该特性。当前的实现为每个创建的ResourceProfile获取新的执行程序，目前必须是精确匹配的。Spark不会尝试将需要不同于执行器创建时所使用的ResourceProfile的任务放入执行器中。未被使用的执行程序将使用动态分配逻辑闲置超时。该特性的默认配置是每个阶段只允许一个ResourceProfile。如果用户把RDD关联的ResourceProfile超过1个，Spark默认会抛出异常。参见config spark.scheduler.resource.profilemergeneconflicts来控制该行为。Spark在Spark.scheduler.resource. profilemergeneconflicts启用时实现的当前合并策略是冲突的ResourceProfiles中每个资源的简单最大值。Spark将使用每个资源的最大值创建一个新的ResourceProfile。

基于push的shuffle概述

基于push的shuffle有助于提高spark shuffle的可靠性和性能。它将map任务生成的shuffle块推到远程外部shuffle服务，以便每个shuffle分区合并。Reduce任务获取合并的shuffle分区和原始shuffle块的组合作为它们的输入数据，从而将外部shuffle服务的小的随机读取磁盘转换为大的顺序读取。为reduce任务提供更好的数据位置的可能性还有助于最小化网络IO。在某些情况下，基于push的shuffle优先于批处理获取，比如合并输出可用时的分区合并。

基于push的shuffle提高了长时间运行的作业/查询的性能，这些作业/查询在shuffle期间涉及大量磁盘I/O。目前，它不太适合快速运行的作业/查询，处理少量的shuffle数据。这将在未来的版本中得到进一步的改进。