Spark Configuration配置

2024年3月1日 114次阅读 来源: spark

Spark可以通过三种方式配置系统:

  • 通过SparkConf对象, 或者Java系统属性配置Spark的应用参数
  • 通过每个节点上的conf/spark-env.sh脚本为每台机器配置环境变量
  • 通过log4j.properties配置日志属性

Spark属性

Spark属性可以为每个应用分别进行配置，这些属性可以直接通过SparkConf设定，也可以通过set方法设定相关属性。 
下面展示了在本地机使用两个线程并发执行的配置代码：

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

对于部分时间参数需要制定单位，例如

  • 时间单位：ms、s、m（min）、h、d、y分别表示毫秒、秒、分钟、小时、天和年。
  • 存储单位：
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，并在调用脚本时制定相关参数

./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提供两种方式动态加载配置

  • 命令行参数动态设定，例如–conf –master
  • 通过配置文件。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

参数设置在执行时会进行合并，默认最高优先级是通过代码设置，其次是通过命令行参数，最后是默认的配置文件。

查看Spark配置

可以通过web界面http://:4040中的Environment查看Spark配置信息（仅显示spark-defaults.conf、SparkConf和命令行参数）。可以根据web页面确定配置属性是否生效。

配置参数 Available Properties

大部分配置参数都有默认值，以下是常用配置

Application Properties

属性默认值描述
spark.app.name(none)应用程序的名称，会在日志和webUI显示
spark.driver.cores1driver程序占用的CPU核数，只在cluster模式下有小。
spark.driver.maxResultSize1g对Spark每个action结果集大小的限制，最少是1M，若设为0则不限制大小。若Job结果超过限制则会异常退出，若结果集限制过大也可能造成OOM问题。
spark.driver.memory1gdriver进程可用的内存。注意：不能在代码中配置，因为此时driver已经启动，可以通过–driver-memory命令行参数或者配置文件进行配置。
spark.executor.memory1g每个executor可用的内存数量 (e.g. 2g, 8g).
spark.extraListeners(none)一系列实现SparkListener的类，spark监听总线会创建这些类的实例。
spark.local.dir/tmp用于存储mpp输出文件和RDD缓存文件，常配置在SSD等存储设备上，可以通过逗号分隔指定多个目录。 注意: 在Spark 1.0 后续版本，会被SPARK_LOCAL_DIRS (Standalone, Mesos) or LOCAL_DIRS (YARN) 环境变量覆盖.
spark.logConffalse将SparkConf 的有效配置作为INFO进行记录
spark.master(none)集群master节点

 

运行时环境

属性默认值描述
spark.driver.userClassPathFirstfalse用户指定的jars优先于Spark的库。用于解决用户与环境的版本冲突
spark.executor.logs.rolling.maxRetainedFiles（none）系统保留日志的最大数量，当超限时，旧的日志被删除，默认不启动
spark.executor.logs.rolling.time.intervaldaily设置日志rolling时间间隔，默认rolling不启动
spark.executor.userClassPathFirstfalseexecutor执行时，用户指定的jars优先于Spark的库。用于解决用户与环境的版本冲突
spark.python.worker.memory512m每个worker进程在聚集时的内存上限，若超限则输出到硬盘

 

 

 

 

 

Shuffle 行为

属性默认值描述
spark.reducer.maxSizeInFlight48m多个reduce任务从map输出获取结果的最大尺寸。由于每个reducer需要创建缓存保留数据，除非内存很大，一般不要修改此参数
spark.shuffle.compresstrue是否对map的输出结果进行压缩，压缩器为spark.io.compression.codec
spark.shuffle.file.buffer32k每个shuffle文件输出流的内存缓存区大小。这些缓冲区减少了系统IO的调用次数
spark.shuffle.managersortshuffle数据的实现方法，包括sort和hash两种。sort内存利用率更改，从1.2版本后sort作为默认实现方法
spark.shuffle.service.enabledfalse激活外部shuffle服务。服务维护executor写的文件，因而executor可以被安全移除。需要设置spark.dynamicAllocation.enabled 为true，同事指定外部shuffle服务。
spark.shuffle.service.port7337默认的外部shuffle服务端口
spark.shuffle.sort.bypassMergeThreshold200用于设置在Reducer的partition数目少于多少的时候，Sort Based Shuffle内部不使用Merge Sort的方式处理数据，而是直接将每个partition写入单独的文件。这个方式和Hash Based的方式是类似的，区别就是在最后这些文件还是会合并成一个单独的文件，并通过一个index索引文件来标记不同partition的位置信息。从Reducer看来，数据文件和索引文件的格式和内部是否做过Merge Sort是完全相同的。这个可以看做SortBased Shuffle在Shuffle量比较小的时候对于Hash Based Shuffle的一种折衷。当然了它和Hash Based Shuffle一样，也存在同时打开文件过多导致内存占用增加的问题。因此如果GC比较严重或者内存比较紧张，可以适当的降低这个值。
spark.shuffle.spill.compresstrue若为true，代表处理的中间结果在spill到本地硬盘时都会进行压缩，在将中间结果取回进行merge的时候，要进行解压。在Disk IO成为瓶颈的场景下，这个被设置为true可能比较合适；如果本地硬盘是SSD，那么这个设置为false可能比较合适。

 

Spark UI

属性默认值描述
spark.eventLog.compressfalse是否压缩事务日志，当spark.eventLog.enabled为true时有效
spark.eventLog.dirfile:///tmp/spark-events记录event日志的目录，也可以是hdfs目录
spark.eventLog.enabledfalse是否记录Spark events，用于在应用执行完后重建Web UI
spark.eventLog.enabledtrue是否允许通过web UI kill掉stages和相关的job
spark.ui.port4040应用统计信息的端口
spark.ui.retainedJobs1000spark UI和status APIs在垃圾回收之前记录的任务数
spark.ui.retainedStages1000spark UI和status APIs在垃圾回收之前记录的Stage数
spark.worker.ui.retainedExecutors1000spark UI和status APIs在垃圾回收之前记录的executor数目
spark.worker.ui.retainedDrivers1000同上
spark.worker.ui.retainedExecutions1000同上
spark.worker.ui.retainedBatches1000同上

 

 

 

 

 

 

 

 

 

 

压缩和序列化

属性默认值描述
spark.broadcast.compresstrue广播变量是否被压缩
spark.closure.serializerorg.apache.spark.serializer.JavaSerializer闭包的序列化类，目前只支持java序列化
spark.io.compression.codecsnappy内部数据RDD的压缩编码器，用于RDD、广播变量和shuffle数据压缩。支持三种编码器：lz4、lzf和snappy。
spark.io.compression.lz4.blockSize32k压缩块大小
spark.io.compression.snappy.blockSize32k压缩块大小
spark.kryo.classesToRegister（none）若使用kryo序列化，本参数指定需要注册的自定义类
spark.kryo.referenceTrackingtrue(false when using Spark SQL Thrift Server)序列化时是否使用相同的对象，若对象图谱中包含同一对象的多个副本，会提高性能。若不存在该情况，关闭可以提高性能
spark.kryo.registrationRequiredfalse是否需要kry注册，若为true，在序列化未注册的类时kryo会抛出异常；若为false，对于未注册的类，kryo会在每个对象写入类名，降低了性能。
spark.kryo.registrator（none）指定自定义的kryo注册类
spark.kryoserializer.buffer.max64mkryo序列化的缓冲区大小，需要比所有序列化对象大
spark.kryoserializer.buffer64k初始化的序列化缓冲区，可以根据需要增长到spark.kryoserializer.buffer.max
spark.rdd.compressfalse是否序列化RDD分区，能通过耗费大量CPU降低存储空间
spark.serializerorg.apache.spark.serializer.JavaSerializer序列化对象的类，建议使用org.apache.spark.serializer.KryoSerializer
spark.serializer.objectStreamReset100当序列化对象时，为了减少IO会缓存大量数据，然而这会阻止垃圾回收，可以通过reset将刷新缓冲区。

 

内存管理

属性默认值描述
spark.memory.fraction0.75用于执行和存储的内存比例。值越小，计算内存越小，缓冲区数据被排除的可能越大。这个比例剩余的部分用于存储spark元数据、用户数据结构，最好使用默认值。
spark.memory.storageFraction0.5在存储和计算内存中，缓存所占的内存比例，值越大，计算可用内存越少。
spark.memory.offHeap.enabledfalse若为true，则spark会尝试使用堆外内存，同时要求spark.memory.offHeap.size必须是正数
spark.memory.offHeap.size0堆外内存可用的字节数
spark.memory.useLegacyModefalse是否可以使用传统内存管理。本参数为true时，以下参数（已废弃）才有效：spark.shuffle.memoryFraction spark.storage.memoryFraction spark.storage.unrollFraction

 

执行操作

属性默认值描述
spark.broadcast.blockSize4mTorrentBroadcastFactory中每个block的大小。若值太大会降低广播的并行度，若值太小则可能出现BlockManager瓶颈
spark.broadcast.factoryorg.apache.spark.broadcast.TorrentBroadcastFactory广播的实现
spark.cleaner.ttl（infinite）spark记忆元数据的时间。若超时则清理，用于长时间运行例如spark stream应用，需要注意被缓存的RDD超时也会被清理。
spark.executor.cores在Yarn是1；standalone中是所有可用的core每个executor可用的CPU核心数目，standalone模式下，每个worker会每个executor使用一个CPU核心
spark.default.parallelism对于reduceByKey和join操作，是RDD中最大分区数；对于parallelize操作，分区数与集群管理相关：本地模式（CPU核心数作为分区数）、Mesos（8）、其他（所有执行器的核心数与2求最大值）默认的并行数
spark.executor.heartbeatInterval10sexecutor与driver的心跳间隔
spark.files.fetchTimeout60sSparkContext.addFile的超时值
spark.files.useFetchCachetrue若为true，同一应用的执行器间通过局部缓存优化；若为false则各个executor获取各自文件
spark.files.overwritefalse当目标文件存在时是否重写
spark.hadoop.cloneConffalse若为true，则为每个task拷贝hadoop的配置对象；
spark.hadoop.validateOutputSpecstrue若设置为true，saveAsHadoopFile会验证输出目录是否存在。虽然设为false可以忽略文件存在的异常，但建议使用Hadoop文件系统的API手动删除输出目录。当通过Spark Streaming的StreamingContext时本参数会被忽略，因为当进行checkpoint恢复时会重写已经存在的文件。
spark.storage.memoryMapThreshold2m读取文件块时Spark内存map最小的大小。当map所占内存接近或小于操作系统page大小时，内存映射负载很大
spark.externalBlockStore.blockManagerorg.apache.spark.storage.TachyonBlockManager存储RDDs的外部文件块管理器。文件系统的URL被设置为spark.externalBlockStore.url
spark.externalBlockStore.baseDirSystem.getProperty(“java.io.tmpdir”)外部块存储RDD的目录。文件系统URL被设置为spark.externalBlockStore.url
spark.externalBlockStore.urltachyon://localhost:19998 for Tachyon代表外部块文件系统的URL

 

网络

属性默认值描述
spark.akka.frameSize128最大消息大小（MB）。一般用于限制executor与driver之间的信息大小，若运行几千个map和reduce任务，可以适当调大参数
spark.akka.heartbeat.interval1000s可以调成很大，用于禁止Akka内部的传输失败检测。越大的时间间隔减少网络负载，越小的间隔容易进行Akka错误检测。
spark.akka.heartbeat.pauses6000s与spark.akka.heartbeat.interval类似
spark.akka.threads4actor用于传输的线程个数。当driver有较多CPU是可以调大该值
spark.akka.timeout100sspark节点间沟通的超时时间
spark.blockManager.portrandomblock 管理器的监听端口
spark.broadcast.portrandomdriver的http广播监听端口
spark.driver.host（local hostname)driver监听的主机名或者IP地址。用于master和executor的信息传输
spark.driver.portrandomdriver监听的接口
spark.executor.portrandomexecutor监听的端口
spark.fileserver.portrandomdriver 文件服务监听的接口
spark.network.timeout120s默认所有网络交互的超时时间
spark.port.maxRetries16端口重试连接最大次数
spark.replClassServer.portrandomdriver类服务监听的接口
spark.rpc.numRetries3RPC任务重试的次数
spark.rpc.retry.wait3sRPC任务ask操作的延时
spark.rpc.askTimeout120sRPC任务等待超时
spark.rpc.lookupTimeout120sRPC远程lookup操作超时时间

 

 作业调度Scheduling

属性默认值描述
spark.cores.max（not set)spark应用可用最大CPU内核数，若未设置，stanalone集群使用 spark.deploy.defaultCores作为参数，Mesos可以使用所有CPU。
spark.locality.wait3sdata-local或less-local任务启动任务超时时间。若任务时间长且数据不再本地，则最后调大
spark.locality.wait.nodespark.locality.waitCustomize the locality wait for node locality. For example, you can set this to 0 to skip node locality and search immediately for rack locality (if your cluster has rack information).
spark.locality.wait.processspark.locality.waitCustomize the locality wait for process locality. This affects tasks that attempt to access cached data in a particular executor process.
spark.locality.wait.rackspark.locality.waitCustomize the locality wait for rack locality.
spark.scheduler.maxRegisteredResourcesWaitingTime30sMaximum amount of time to wait for resources to register before scheduling begins.
spark.scheduler.modeFIFO作业调度模式。可以设置为FAIR公平调度或者FIFO先进先出
spark.scheduler.revive.interval1sThe interval length for the scheduler to revive the worker resource offers to run tasks.
spark.speculationfalse若设置为true，则会根据执行慢的stage多次启动，以最先完成为准。
spark.speculation.interval100msspeculate 的频率
spark.speculation.multiplier1.5当task比所有任务执行时间的中值长多少倍时启动speculate
spark.speculation.quantile0.75启动speculate前任务完成数据量所占比例值
spark.task.cpus1每个task分配的cpu数量
spark.task.maxFailures4task失败的最多次数，比重试次数多1

 

动态分配内存

属性默认值描述
spark.dynamicAllocation.enabledfalse是否启动动态资源分配
spark.dynamicAllocation.executorIdleTimeout60s若动态分配设为true且executor处于idle状态的时间已超时，则移除executor
spark.dynamicAllocation.cachedExecutorIdleTimeoutinfinity若executor缓存数据超时，且动态内存分配为true，则移除缓存
spark.dynamicAllocation.initialExecutorsspark.dynamicAllocation.minExecutors若动态分配为true，执行器的初始数量
spark.dynamicAllocation.maxExecutorinfinity执行器最大数量
spark.dynamicAllocation.minExecutor0执行器最少数量
spark.dynamicAllocation.schedulerBacklogTimeout1sIf dynamic allocation is enabled and there have been pending tasks backlogged for more than this duration, new executors will be requested.
spark.dynamicAllocation.sustainedSchedulerBacklogTimeoutschedulerBacklogTimeoutSame as spark.dynamicAllocation.schedulerBacklogTimeout, but used only for subsequent executor requests.

 

安全

属性默认值描述
spark.acls.enablefalseWhether Spark acls should are enabled. If enabled, this checks to see if the user has access permissions to view or modify the job. Note this requires the user to be known, so if the user comes across as null no checks are done. Filters can be used with the UI to authenticate and set the user.
spark.admin.aclsEmptyComma separated list of users/administrators that have view and modify access to all Spark jobs. This can be used if you run on a shared cluster and have a set of administrators or devs who help debug when things work. Putting a “*” in the list means any user can have the priviledge of admin.
spark.authenticatefalseWhether Spark authenticates its internal connections. See spark.authenticate.secret if not running on YARN.
spark.authenticate.secretNoneSet the secret key used for Spark to authenticate between components. This needs to be set if not running on YARN and authentication is enabled.
spark.authenticate.enableSaslEncryptionfalseEnable encrypted communication when authentication is enabled. This option is currently only supported by the block transfer service.
spark.network.sasl.serverAlwaysEncryptfalseDisable unencrypted connections for services that support SASL authentication. This is currently supported by the external shuffle service.
spark.core.connection.ack.wait.timeout60sHow long for the connection to wait for ack to occur before timing out and giving up. To avoid unwilling timeout caused by long pause like GC, you can set larger value.
spark.core.connection.auth.wait.timeout30sHow long for the connection to wait for authentication to occur before timing out and giving up.
spark.modify.aclsEmptyComma separated list of users that have modify access to the Spark job. By default only the user that started the Spark job has access to modify it (kill it for example). Putting a “*” in the list means any user can have access to modify it.
spark.ui.filtersNoneComma separated list of filter class names to apply to the Spark web UI. The filter should be a standard javax servlet Filter. Parameters to each filter can also be specified by setting a java system property of:spark..params=’param1=value1,param2=value2’For example:-Dspark.ui.filters=com.test.filter1 -Dspark.com.test.filter1.params=’param1=foo,param2=testing’
spark.ui.view.aclsEmptyComma separated list of users that have view access to the Spark web ui. By default only the user that started the Spark job has view access. Putting a “*” in the list means any user can have view access to this Spark job.

 

加密Encryption

属性默认值描述
spark.ssl.enabledfalse是否在所有协议中支持SSL连接
spark.ssl.enabledAlgorithmsEmpty一些列的密码运算，指定的cipher需要被JVM支持
spark.ssl.keyPasswordNone私钥密码
spark.ssl.keyStoreNonekey存储文件，可以是组件启动的相对路径也可以是绝对路径
spark.ssl.keyStorePasswordNoneA password to the key-store
spark.ssl.protocolNoneA protocol name. The protocol must be supported by JVM. The reference list of protocols one can find on this page.
spark.ssl.trustStoreNoneA path to a trust-store file. The path can be absolute or relative to the directory where the component is started in.
spark.ssl.trustStorePasswordNoneA password to the trust-store.

 

 

 

 

 

 

 

Spark Streaming

属性默认值描述
spark.streaming.backpressure.enabledfalseEnables or disables Spark Streaming’s internal backpressure mechanism (since 1.5). This enables the Spark Streaming to control the receiving rate based on the current batch scheduling delays and processing times so that the system receives only as fast as the system can process. Internally, this dynamically sets the maximum receiving rate of receivers. This rate is upper bounded by the values spark.streaming.receiver.maxRate and spark.streaming.kafka.maxRatePerPartition if they are set (see below).
spark.streaming.blockInterval200msInterval at which data received by Spark Streaming receivers is chunked into blocks of data before storing them in Spark. Minimum recommended – 50 ms. See the performance tuning section in the Spark Streaming programing guide for more details.
spark.streaming.receiver.maxRatenot setMaximum rate (number of records per second) at which each receiver will receive data. Effectively, each stream will consume at most this number of records per second. Setting this configuration to 0 or a negative number will put no limit on the rate. See the deployment guide in the Spark Streaming programing guide for mode details.
spark.streaming.receiver.writeAheadLog.enablefalseEnable write ahead logs for receivers. All the input data received through receivers will be saved to write ahead logs that will allow it to be recovered after driver failures. See the deployment guide in the Spark Streaming programing guide for more details.
spark.streaming.unpersisttrueForce RDDs generated and persisted by Spark Streaming to be automatically unpersisted from Spark’s memory. The raw input data received by Spark Streaming is also automatically cleared. Setting this to false will allow the raw data and persisted RDDs to be accessible outside the streaming application as they will not be cleared automatically. But it comes at the cost of higher memory usage in Spark.
spark.streaming.stopGracefullyOnShutdownfalseIf true, Spark shuts down the StreamingContext gracefully on JVM shutdown rather than immediately.
spark.streaming.kafka.maxRatePerPartitionnot setMaximum rate (number of records per second) at which data will be read from each Kafka partition when using the new Kafka direct stream API. See the Kafka Integration guide for more details.
spark.streaming.kafka.maxRetries1Maximum number of consecutive retries the driver will make in order to find the latest offsets on the leader of each partition (a default value of 1 means that the driver will make a maximum of 2 attempts). Only applies to the new Kafka direct stream API.
spark.streaming.ui.retainedBatches1000How many batches the Spark Streaming UI and status APIs remember before garbage collecting.
spark.streaming.driver.writeAheadLog.closeFileAfterWritefalseWhether to close the file after writing a write ahead log record on the driver. Set this to ‘true’ when you want to use S3 (or any file system that does not support flushing) for the metadata WAL on the driver.
spark.streaming.receiver.writeAheadLog.closeFileAfterWritefalseWhether to close the file after writing a write ahead log record on the receivers. Set this to ‘true’ when you want to use S3 (or any file system that does not support flushing) for the data WAL on the receivers.

 

SparkR

属性默认值描述
spark.r.numRBackendThreads2RBackend维护的RPC句柄个数
spark.r.commandRscriptExecutable for executing R scripts in cluster modes for both driver and workers.
spark.r.driver.commandspark.r.commandExecutable for executing R scripts in client modes for driver. Ignored in cluster modes

 

 

 

 

其他参数具体参见https://spark.apache.org/docs/latest/configuration.html

环境变量

部分Spark设置可以通过配置环境变量（在conf/spark-env.sh中设置）实现。在standalone和Mesos模式中，这个文件可以设定机器特定的信息，例如主机名。由于spark-env.sh安装后并不存在，可以拷贝spark-env.sh.template，并确保它有执行权限。 
以下是spark-env.sh的常用参数：

环境变量描述
JAVA_HOMEjava安装目录
PYSPARK_PYTHON运行pyspark的python可执行文件，默认是python2.7
SPARK_DRIVER_RSparkR shell的R可执行文件，默认是R
SPARK_LOCAL_IP机器绑定的IP地址
SPARK_PUBLIC_DNSSpark程序向外广播的主机名

 

 

 

 

 

 

除此之外还有一些spark standalone集群设置的参数，例如每个机器运行的最大内存、CPU核数等。

配置日志

Spark使用log4j记录日志。可以通过配置conf/log4j.properties文件配置日志。

覆盖配置目录

通过指定SPARK_CONF_DIR变量，可以覆盖默认的SAPRK_HOME/conf下面的配置，例如spark-defaults.conf, spark-env.sh, log4j.properties等待。

集成Hadoop集群配置

若想通过spark读写HDFS，需要将以下两个配置文件拷贝到spark classpath目录下 
+ hdfs-size.xml ：提供HDFS客户端默认的操作 
+ core-site.xml ：设置默认的文件系统名称

虽然不同发行版本配置文件不同，但一般都在/etc/hadoop/conf目录下。为了使得spark可以找到这些配置文件，在spark-env.sh文件中配置HADOOP_CONF_DIR变量。

    原文作者：spark
    原文地址: https://www.cnblogs.com/sh425/p/7384467.html
    本文转自网络文章，转载此文章仅为分享知识，如有侵权，请联系博主进行删除。
点赞