From 3de1a5904c7e273cf1bf27d3b634cd141c42b1d3 Mon Sep 17 00:00:00 2001 From: CodingCat Date: Tue, 4 Mar 2014 23:23:21 -0500 Subject: [PATCH 01/33] spark.deploy.retainedApplications --- docs/configuration.md | 31 ++++++++++++++++++++----------- 1 file changed, 20 insertions(+), 11 deletions(-) diff --git a/docs/configuration.md b/docs/configuration.md index dc5553f3da770..1e7b087425eb3 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -454,17 +454,6 @@ Apart from these, the following properties are also available, and may be useful Log the supplied SparkConf as INFO at start of spark context. - - spark.deploy.spreadOut - true - - Whether the standalone cluster manager should spread applications out across nodes or try - to consolidate them onto as few nodes as possible. Spreading out is usually better for - data locality in HDFS, but consolidating is more efficient for compute-intensive workloads.
- Note: this setting needs to be configured in the standalone cluster master, not in individual - applications; you can set it through SPARK_JAVA_OPTS in spark-env.sh. - - spark.deploy.defaultCores (infinite) @@ -478,6 +467,26 @@ Apart from these, the following properties are also available, and may be useful applications; you can set it through SPARK_JAVA_OPTS in spark-env.sh. + + spark.deploy.retainedApplications + 200 + + The maximum number of records on completed applications kept in the Master node. When the completed + application number exceeds the threshold, the first Max(spark.deploy.retainedApplications, 1) will be + discarded. + + + + spark.deploy.spreadOut + true + + Whether the standalone cluster manager should spread applications out across nodes or try + to consolidate them onto as few nodes as possible. Spreading out is usually better for + data locality in HDFS, but consolidating is more efficient for compute-intensive workloads.
+ Note: this setting needs to be configured in the standalone cluster master, not in individual + applications; you can set it through SPARK_JAVA_OPTS in spark-env.sh. + + spark.files.overwrite false From 003450050f7b4de9f45ed01578277e200d7affe6 Mon Sep 17 00:00:00 2001 From: CodingCat Date: Tue, 4 Mar 2014 23:30:55 -0500 Subject: [PATCH 02/33] spark.dead.worker.persistence --- docs/configuration.md | 11 ++++++++++- 1 file changed, 10 insertions(+), 1 deletion(-) diff --git a/docs/configuration.md b/docs/configuration.md index 1e7b087425eb3..91a73e490c4c6 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -454,6 +454,15 @@ Apart from these, the following properties are also available, and may be useful Log the supplied SparkConf as INFO at start of spark context. + + spark.dead.worker.persistence + 15 + + This parameter controls when Spark will remove the worker information from the UI if it is dead. + If it has been longer than (spark.dead.worker.persistence + 1) * spark.worker.timeout + seconds since the worker is dead, Spark will remove it from UI. + + spark.deploy.defaultCores (infinite) @@ -465,7 +474,7 @@ Apart from these, the following properties are also available, and may be useful the whole cluster by default.
Note: this setting needs to be configured in the standalone cluster master, not in individual applications; you can set it through SPARK_JAVA_OPTS in spark-env.sh. - + spark.deploy.retainedApplications From 7432af47eb9043eaacc64c1d906c6dae17d9f805 Mon Sep 17 00:00:00 2001 From: CodingCat Date: Tue, 4 Mar 2014 23:36:55 -0500 Subject: [PATCH 03/33] spark.deploy.recoveryDirectory - Master.scala L51 --- docs/configuration.md | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/docs/configuration.md b/docs/configuration.md index 91a73e490c4c6..5d07cf9c5a015 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -476,6 +476,13 @@ Apart from these, the following properties are also available, and may be useful applications; you can set it through SPARK_JAVA_OPTS in spark-env.sh. + + spark.deploy.recoveryDirectory + "" + + The user can specify the directory for the master recovery file when using FileSystemPersistenceEngine. + + spark.deploy.retainedApplications 200 From 679ee116e19ebf69180086d0559d938699eff1d0 Mon Sep 17 00:00:00 2001 From: CodingCat Date: Wed, 5 Mar 2014 00:00:09 -0500 Subject: [PATCH 04/33] spark.deploy.zookeeper.url - SparkCuratorUtil.scala L34 --- docs/configuration.md | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/docs/configuration.md b/docs/configuration.md index 5d07cf9c5a015..e01e31d2c6ac2 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -503,6 +503,13 @@ Apart from these, the following properties are also available, and may be useful applications; you can set it through SPARK_JAVA_OPTS in spark-env.sh. + + spark.deploy.zookeeper.url + (not set) + + The URL of zookeeper cluster. The user has to set this parameter explicitly when using ZooKeeper. + + spark.files.overwrite false From c22d52015fd372cb30913d50ddeb439bfcbf5dae Mon Sep 17 00:00:00 2001 From: CodingCat Date: Wed, 5 Mar 2014 00:31:41 -0500 Subject: [PATCH 05/33] spark.repl.class.uri - Executor.scala 313 --- docs/configuration.md | 7 +++++++ .../scala/org/apache/spark/repl/ExecutorClassLoader.scala | 2 +- 2 files changed, 8 insertions(+), 1 deletion(-) diff --git a/docs/configuration.md b/docs/configuration.md index e01e31d2c6ac2..4775df187a3cd 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -217,6 +217,13 @@ Apart from these, the following properties are also available, and may be useful per reduce task, so keep it small unless you have a large amount of memory. + + spark.repl.class.uri + null + + This parameter specifies the URI of the class defined by REPL as user-defined code. + + spark.closure.serializer org.apache.spark.serializer.
JavaSerializer diff --git a/repl/src/main/scala/org/apache/spark/repl/ExecutorClassLoader.scala b/repl/src/main/scala/org/apache/spark/repl/ExecutorClassLoader.scala index e3bcf7f30ac8d..cdfe8abdd6f44 100644 --- a/repl/src/main/scala/org/apache/spark/repl/ExecutorClassLoader.scala +++ b/repl/src/main/scala/org/apache/spark/repl/ExecutorClassLoader.scala @@ -33,7 +33,7 @@ import org.objectweb.asm.Opcodes._ * used to load classes defined by the interpreter when the REPL is used */ class ExecutorClassLoader(classUri: String, parent: ClassLoader) -extends ClassLoader(parent) { + extends ClassLoader(parent) { val uri = new URI(classUri) val directory = uri.getPath From a0e842d13baf62609fe0b4d04890d9d916aff166 Mon Sep 17 00:00:00 2001 From: CodingCat Date: Wed, 5 Mar 2014 09:13:25 -0500 Subject: [PATCH 06/33] spark.core.connection.*.threads.* - ConnectionManager.scala --- docs/configuration.md | 69 +++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 69 insertions(+) diff --git a/docs/configuration.md b/docs/configuration.md index 4775df187a3cd..af02993357561 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -87,6 +87,75 @@ there are at least five properties that you will commonly want to control: infinite (all available cores) on Mesos. + + spark.core.connection.handler.threads.min + 20 + + The number of threads to keep in the thread pool handling acceptance of the new connection, even they are + idle. + + + + spark.core.connection.handler.threads.max + 60 + + The maximum number of threads to allow in the pool handling acceptance of the new connection. + + + + spark.core.connection.handler.threads.keepalive + 60 + + When the number of threads in the pool handling acceptance of the new connection is greater than the core, + this is the maximum time that excess idle threads will wait for new tasks before terminating. + + + + spark.core.connection.io.threads.min + 20 + + The number of threads to keep in the thread pool handling BufferMessage receive/send, even they are + idle. + + + + spark.core.connection.io.threads.max + 60 + + The maximum number of threads to allow in the pool handling BufferMessage receive/send. + + + + spark.core.connection.io.threads.keepalive + 60 + + When the number of threads in the pool handling BufferMessage receive/send is greater than the core, + this is the maximum time that excess idle threads will wait for new tasks before terminating. + + + + spark.core.connection.connect.threads.min + 20 + + The number of threads to keep in the thread pool handling the establishment of the connection, even they are + idle. + + + + spark.core.connection.connect.threads.max + 60 + + The maximum number of threads to allow in the pool handling the establishment of the connection. + + + + spark.core.connection.connect.threads.keepalive + 60 + + When the number of threads in the pool handling the establishment of the connection is greater than the core, + this is the maximum time that excess idle threads will wait for new tasks before terminating. + + From 9acaad4acc12566d1122114334a7b848f7d81e0e Mon Sep 17 00:00:00 2001 From: CodingCat Date: Wed, 5 Mar 2014 09:21:28 -0500 Subject: [PATCH 07/33] spark.shuffle.netty.connect.timeout - ShuffleCopier.scala L38 --- docs/configuration.md | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/docs/configuration.md b/docs/configuration.md index af02993357561..35d6007cfcfc1 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -487,6 +487,13 @@ Apart from these, the following properties are also available, and may be useful reduce the number of disk seeks and system calls made in creating intermediate shuffle files. + + spark.shuffle.netty.connect.timeout + 60000 + + This parameter sets the timeout threshold for the Netty connection when shuffling. (in milliseconds) + + spark.shuffle.spill true From 02826441047be8c3cd74fbf5bdaf438c9f3be128 Mon Sep 17 00:00:00 2001 From: CodingCat Date: Wed, 5 Mar 2014 09:50:05 -0500 Subject: [PATCH 08/33] spark.mesos.extra.cores - CoarseMesosSchedulerBackend L79 --- docs/configuration.md | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/docs/configuration.md b/docs/configuration.md index 35d6007cfcfc1..3b9c1418d896a 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -202,6 +202,14 @@ Apart from these, the following properties are also available, and may be useful duration of the Spark job. + + spark.mesos.extra.cores + 0 + + When using Mesos, the user can specify how many cores each task can use in addition to the cores it gets from + the resource offer. + + spark.ui.port 4040 From e877298a3b3cc5c784bccb34eee2019b7a01ed3b Mon Sep 17 00:00:00 2001 From: CodingCat Date: Wed, 5 Mar 2014 09:54:49 -0500 Subject: [PATCH 09/33] spark.scheduler.allocation.file - SchedulableBuilder.scala L55 --- docs/configuration.md | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/docs/configuration.md b/docs/configuration.md index 3b9c1418d896a..033eeacecf3f3 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -278,6 +278,14 @@ Apart from these, the following properties are also available, and may be useful multi-user services. + + spark.scheduler.allocation.file + (not set) + + User can specify the allocation file location for the FAIR scheduler with this parameter. By default, + Spark uses "fairscheduler.xml" in conf directory as the allocation file. + + spark.scheduler.revive.interval 1000 From c81bfa8a585e389a10294cb6cbe8cea597aa7960 Mon Sep 17 00:00:00 2001 From: CodingCat Date: Wed, 5 Mar 2014 10:05:30 -0500 Subject: [PATCH 10/33] spark.resultGetter.threads - TaskResultGetter L33 --- docs/configuration.md | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/docs/configuration.md b/docs/configuration.md index 033eeacecf3f3..d94555997841b 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -302,6 +302,14 @@ Apart from these, the following properties are also available, and may be useful per reduce task, so keep it small unless you have a large amount of memory. + + spark.resultGetter.threads + 4 + + Set the size of thread pool in TaskScheduler to deserialize and get the task results from remote end when necessary. + The size of the thread pool is fixed. + + spark.repl.class.uri null From a8618d3afc34ce67df8f6575db1536e467f5e509 Mon Sep 17 00:00:00 2001 From: CodingCat Date: Wed, 5 Mar 2014 10:10:32 -0500 Subject: [PATCH 11/33] spark.starvation.timeout - TaskSchedulerImpl L63 --- docs/configuration.md | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/docs/configuration.md b/docs/configuration.md index d94555997841b..ec2d18dfb884d 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -463,6 +463,13 @@ Apart from these, the following properties are also available, and may be useful applications). Note that any RDD that persists in memory for more than this duration will be cleared as well. + + spark.starvation.timeout + 15000 + + Threshold above which we warn user initial TaskSet may be starved. (in milliseconds) + + spark.streaming.blockInterval 200 From 319cf4835bd3e8f3ab02ec65c27cf1561c42a7b7 Mon Sep 17 00:00:00 2001 From: CodingCat Date: Wed, 5 Mar 2014 10:27:21 -0500 Subject: [PATCH 12/33] spark.task.cpus - L60 --- docs/configuration.md | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/docs/configuration.md b/docs/configuration.md index ec2d18dfb884d..aa6ca93155129 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -494,6 +494,13 @@ Apart from these, the following properties are also available, and may be useful Should be greater than or equal to 1. Number of allowed retries = this value - 1. + + spark.task.cpus + 1 + + CPUs to request per task + + spark.broadcast.blockSize 4096 From 0c371363b78a3395f0940e53758d6b67b8b12d2a Mon Sep 17 00:00:00 2001 From: CodingCat Date: Wed, 5 Mar 2014 10:33:43 -0500 Subject: [PATCH 13/33] spark.logging.exceptionPrintInterval - TaskSetManager.scala L127 --- docs/configuration.md | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/docs/configuration.md b/docs/configuration.md index aa6ca93155129..4ebda3341f52b 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -575,6 +575,13 @@ Apart from these, the following properties are also available, and may be useful Log the supplied SparkConf as INFO at start of spark context. + + spark.logging.exceptionPrintInterval + 10000 + + How frequently the driver reprints duplicate exceptions in full. (in milliseconds) + + spark.dead.worker.persistence 15 From 67046956d40eb87c9b56560df144705ff27cf305 Mon Sep 17 00:00:00 2001 From: CodingCat Date: Wed, 5 Mar 2014 10:38:05 -0500 Subject: [PATCH 14/33] spark.jars - SparkContext.scala L122 --- docs/configuration.md | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/docs/configuration.md b/docs/configuration.md index 4ebda3341f52b..838643edf2d1e 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -568,6 +568,14 @@ Apart from these, the following properties are also available, and may be useful How many times slower a task is than the median to be considered for speculation. + + spark.jars + (not set) + + The external jars to be propagated to the executors from the driver. Users can specify multiple jars by separating them + with ",". + + spark.logConf false From 54abac8bad2866ce60f1abdd2e81fb61096d688a Mon Sep 17 00:00:00 2001 From: CodingCat Date: Wed, 5 Mar 2014 11:09:02 -0500 Subject: [PATCH 15/33] spark.shuffle.copier.threads - BlockFetcherIterator L329 --- docs/configuration.md | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/docs/configuration.md b/docs/configuration.md index 838643edf2d1e..3e2c93ff0b86a 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -231,6 +231,14 @@ Apart from these, the following properties are also available, and may be useful Whether to compress map output files. Generally a good idea. + + spark.shuffle.copier.threads + 6 + + When using Netty, this parameter specifies the number of concurrent threads fetching the shuffle results. This parameter + is only valid when spark.shuffle.use.netty is set to true + + spark.shuffle.spill.compress true From 6f1850d5a732696121e269e105e1d1295d1a527e Mon Sep 17 00:00:00 2001 From: CodingCat Date: Wed, 5 Mar 2014 11:13:06 -0500 Subject: [PATCH 16/33] spark.shuffle.use.netty - BlockManager.scala L446 --- docs/configuration.md | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/docs/configuration.md b/docs/configuration.md index 3e2c93ff0b86a..e446f9a5b2d67 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -239,6 +239,13 @@ Apart from these, the following properties are also available, and may be useful is only valid when spark.shuffle.use.netty is set to true + + spark.shuffle.use.netty + false + + If use Netty to fetch map output in shuffle. + + spark.shuffle.spill.compress true From a7151126e171e7b9116b0ead2d5bd2634b0057dd Mon Sep 17 00:00:00 2001 From: CodingCat Date: Wed, 5 Mar 2014 11:22:09 -0500 Subject: [PATCH 17/33] spark.shuffle.sender.port - BlockManager.scala L59 --- docs/configuration.md | 10 +++++++++- 1 file changed, 9 insertions(+), 1 deletion(-) diff --git a/docs/configuration.md b/docs/configuration.md index e446f9a5b2d67..1dbd7fe59656c 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -236,7 +236,7 @@ Apart from these, the following properties are also available, and may be useful 6 When using Netty, this parameter specifies the number of concurrent threads fetching the shuffle results. This parameter - is only valid when spark.shuffle.use.netty is set to true + is only valid when spark.shuffle.use.netty is set to true. @@ -246,6 +246,14 @@ Apart from these, the following properties are also available, and may be useful If use Netty to fetch map output in shuffle. + + spark.shuffle.sender.port + 0 + + When using Netty to send result for shuffle, this parameter specifies the port to use for data transfer. + This parameter is only valid when spark.shuffle.use.netty is set to true. + + spark.shuffle.spill.compress true From 7264fc6d269adeeb9a2aa49ee4be1ba3b8d381b0 Mon Sep 17 00:00:00 2001 From: CodingCat Date: Wed, 5 Mar 2014 16:37:30 -0500 Subject: [PATCH 18/33] spark.shuffle.sync - BlockManager L472 --- docs/configuration.md | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) diff --git a/docs/configuration.md b/docs/configuration.md index 1dbd7fe59656c..fd286262979f3 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -250,10 +250,17 @@ Apart from these, the following properties are also available, and may be useful spark.shuffle.sender.port 0 - When using Netty to send result for shuffle, this parameter specifies the port to use for data transfer. + When using Netty to send result for shuffle, this parameter specifies the port number for data transfer. This parameter is only valid when spark.shuffle.use.netty is set to true. + + spark.shuffle.sync + false + + If this parameter is set to true, Spark will force all outstanding writing for the Map outputs be flushed to the disk when closing the writers. + + spark.shuffle.spill.compress true From 516c78cf01b230c935711f9e7ed271edfe9e5e89 Mon Sep 17 00:00:00 2001 From: CodingCat Date: Wed, 5 Mar 2014 16:47:39 -0500 Subject: [PATCH 19/33] spark.storage.blockManagerTimeoutIntervalMs - BlockManager L874 --- docs/configuration.md | 10 ++++++++++ 1 file changed, 10 insertions(+) diff --git a/docs/configuration.md b/docs/configuration.md index fd286262979f3..57bb718495839 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -171,6 +171,16 @@ Apart from these, the following properties are also available, and may be useful reduceByKey, etc) when not set by user. + + spark.storage.blockManagerTimeoutIntervalMs + 60000 + + This parameter controls two values: First, the interval length for the BlockManagers running on the executors to send hearbeat + to the BlockManager running on the driver. Please note that in Spark the actual frequency is the value set with this parameter divided by 4. + Second, this value also controls the timeout threshold of the slave node, which is spark.storage.blockManagerSlaveTimeoutMs + + spark.storage.blockManagerTimeoutIntervalMs * 3. (In milliseconds) + + spark.storage.memoryFraction 0.6 From 4d4270bed28f9a29b9179e137665ce103df26790 Mon Sep 17 00:00:00 2001 From: CodingCat Date: Wed, 5 Mar 2014 16:54:58 -0500 Subject: [PATCH 20/33] spark.storage.blockManagerSlaveTimeoutMs - BlockManagerMasterActor L53 --- docs/configuration.md | 11 +++++++++-- 1 file changed, 9 insertions(+), 2 deletions(-) diff --git a/docs/configuration.md b/docs/configuration.md index 57bb718495839..19bd78db7b196 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -177,8 +177,15 @@ Apart from these, the following properties are also available, and may be useful This parameter controls two values: First, the interval length for the BlockManagers running on the executors to send hearbeat to the BlockManager running on the driver. Please note that in Spark the actual frequency is the value set with this parameter divided by 4. - Second, this value also controls the timeout threshold of the slave node, which is spark.storage.blockManagerSlaveTimeoutMs + - spark.storage.blockManagerTimeoutIntervalMs * 3. (In milliseconds) + Second, this value also controls the default timeout threshold of the slave node, which is spark.storage.blockManagerTimeoutIntervalMs * 3. + (In milliseconds) + + + + spark.storage.blockManagerSlaveTimeoutMs + spark.storage.blockManagerTimeoutIntervalMs * 3 + + Number of milliseconds after which the BlockManagerMasterActor considers a BlockManager running on the executor lost if it receives no heartbeats. From 105c9ee0a65f7c2f142307b5c5cb9994b931b59e Mon Sep 17 00:00:00 2001 From: CodingCat Date: Wed, 5 Mar 2014 17:17:58 -0500 Subject: [PATCH 21/33] spark.diskStore.subDirectories - DiskBlockManager L41 --- docs/configuration.md | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/docs/configuration.md b/docs/configuration.md index 19bd78db7b196..9749129032123 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -693,6 +693,14 @@ Apart from these, the following properties are also available, and may be useful The URL of zookeeper cluster. The user has to set this parameter explicitly when using ZooKeeper. + + spark.diskStore.subDirectories + 64 + + Specifies how many subdirectories under each local directory. Setting more sub directories under each local directory + can help to avoid the fat inode. + + spark.files.overwrite false From 0b753784773ab7dba18bd6355673ec263fa663a3 Mon Sep 17 00:00:00 2001 From: CodingCat Date: Wed, 5 Mar 2014 19:06:11 -0500 Subject: [PATCH 22/33] spark.akka.batchSize - AkkaUtils.scala L53 --- docs/configuration.md | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/docs/configuration.md b/docs/configuration.md index 9749129032123..9313ae699caa0 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -441,6 +441,14 @@ Apart from these, the following properties are also available, and may be useful receives no heartbeats. + + spark.akka.batchSize + 15 + + The parameter specifies defines the maximum number of messages to be processed per actor + before the thread jumps to the next actor. Set to 1 for as fair as possible. + + spark.akka.frameSize 10 From 163a5b489b4d0a27bb2cdf43d24864ed85934b31 Mon Sep 17 00:00:00 2001 From: CodingCat Date: Wed, 5 Mar 2014 19:12:35 -0500 Subject: [PATCH 23/33] spark.akka.logAkkaConfig - AkkaUtils L61 --- docs/configuration.md | 9 +++++++++ 1 file changed, 9 insertions(+) diff --git a/docs/configuration.md b/docs/configuration.md index 9313ae699caa0..35f93f0c56b4e 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -458,6 +458,15 @@ Apart from these, the following properties are also available, and may be useful (e.g. using collect() on a large dataset). + + spark.akka.logAkkaConfig + false + + This parameter controls three logging behaviours of Akka: 1) Logging the events including: DisassociatedEvent, + AssociatedEvent, AssociationErrorEvent, RemotingListenEvent, RemotingShutdownEvent, RemotingErrorEvent; 2) Logging + the dead letters during running; 3) Logging the dead letters during the shutdown. + + spark.akka.threads 4 From 8fba8cb69061a31bd031986051a2cd70605648d7 Mon Sep 17 00:00:00 2001 From: CodingCat Date: Wed, 5 Mar 2014 19:17:10 -0500 Subject: [PATCH 24/33] spark.akka.askTimeout/lookupTimeout - AkkaUtils L106/111 --- .../scala/org/apache/spark/util/AkkaUtils.scala | 2 +- docs/configuration.md | 14 ++++++++++++++ 2 files changed, 15 insertions(+), 1 deletion(-) diff --git a/core/src/main/scala/org/apache/spark/util/AkkaUtils.scala b/core/src/main/scala/org/apache/spark/util/AkkaUtils.scala index f26ed47e58046..6c9f266b455b3 100644 --- a/core/src/main/scala/org/apache/spark/util/AkkaUtils.scala +++ b/core/src/main/scala/org/apache/spark/util/AkkaUtils.scala @@ -108,6 +108,6 @@ private[spark] object AkkaUtils { /** Returns the default Spark timeout to use for Akka remote actor lookup. */ def lookupTimeout(conf: SparkConf): FiniteDuration = { - Duration.create(conf.get("spark.akka.lookupTimeout", "30").toLong, "seconds") + Duration.create(conf.getLong("spark.akka.lookupTimeout", 30), "seconds") } } diff --git a/docs/configuration.md b/docs/configuration.md index 35f93f0c56b4e..9748e5b45d39c 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -441,6 +441,20 @@ Apart from these, the following properties are also available, and may be useful receives no heartbeats. + + spark.akka.askTimeout + 30 + + This parameter defines the timeout to use for Akka ask operations. (in seconds) + + + + spark.akka.lookupTimeout + 30 + + This parameter defines the timeout to use for Akka lookup operations. (in seconds) + + spark.akka.batchSize 15 From 952f714491c0efae7b56d6ad50f1cefe1ffc9680 Mon Sep 17 00:00:00 2001 From: CodingCat Date: Wed, 5 Mar 2014 20:10:58 -0500 Subject: [PATCH 25/33] remove the parameters documented in other pages --- docs/configuration.md | 14 -------------- 1 file changed, 14 deletions(-) diff --git a/docs/configuration.md b/docs/configuration.md index 9748e5b45d39c..219e17e5c133a 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -690,13 +690,6 @@ Apart from these, the following properties are also available, and may be useful applications; you can set it through SPARK_JAVA_OPTS in spark-env.sh. - - spark.deploy.recoveryDirectory - "" - - The user can specify the directory for the master recovery file when using FileSystemPersistenceEngine. - - spark.deploy.retainedApplications 200 @@ -717,13 +710,6 @@ Apart from these, the following properties are also available, and may be useful applications; you can set it through SPARK_JAVA_OPTS in spark-env.sh. - - spark.deploy.zookeeper.url - (not set) - - The URL of zookeeper cluster. The user has to set this parameter explicitly when using ZooKeeper. - - spark.diskStore.subDirectories 64 From 302b150e73e8fd7fd302e285c9a2adc22a5b0603 Mon Sep 17 00:00:00 2001 From: CodingCat Date: Wed, 5 Mar 2014 20:25:15 -0500 Subject: [PATCH 26/33] remove allocation file as it is documented in other page --- docs/configuration.md | 8 -------- 1 file changed, 8 deletions(-) diff --git a/docs/configuration.md b/docs/configuration.md index 219e17e5c133a..bc793b7af8ad6 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -325,14 +325,6 @@ Apart from these, the following properties are also available, and may be useful multi-user services. - - spark.scheduler.allocation.file - (not set) - - User can specify the allocation file location for the FAIR scheduler with this parameter. By default, - Spark uses "fairscheduler.xml" in conf directory as the allocation file. - - spark.scheduler.revive.interval 1000 From 5aa0a472d1566f0997be1f25fdd1f8acf53aa9b2 Mon Sep 17 00:00:00 2001 From: CodingCat Date: Wed, 5 Mar 2014 23:58:11 -0500 Subject: [PATCH 27/33] Sort the configuration parameters in configuration.md --- docs/configuration.md | 669 +++++++++++++++++++++--------------------- 1 file changed, 334 insertions(+), 335 deletions(-) diff --git a/docs/configuration.md b/docs/configuration.md index bc793b7af8ad6..22bcedd9e4a07 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -39,21 +39,22 @@ there are at least five properties that you will commonly want to control: - - + + - - + + @@ -76,292 +77,308 @@ there are at least five properties that you will commonly want to control: - - + + +
Property NameDefaultMeaning
spark.executor.memory512mspark.cores.max(not set) - Amount of memory to use per executor process, in the same format as JVM memory strings (e.g. 512m, 2g). + When running on a standalone deploy cluster or a + Mesos cluster in "coarse-grained" + sharing mode, the maximum amount of CPU cores to request for the application from + across the cluster (not from each machine). If not set, the default will be + spark.deploy.defaultCores on Spark's standalone cluster manager, or + infinite (all available cores) on Mesos.
spark.serializerorg.apache.spark.serializer.
JavaSerializer		spark.executor.memory512m - Class to use for serializing objects that will be sent over the network or need to be cached - in serialized form. The default of Java serialization works with any Serializable Java object but is - quite slow, so we recommend using org.apache.spark.serializer.KryoSerializer - and configuring Kryo serialization when speed is necessary. Can be any subclass of - org.apache.spark.Serializer. + Amount of memory to use per executor process, in the same format as JVM memory strings (e.g. 512m, 2g).
spark.cores.max(not set)spark.serializerorg.apache.spark.serializer.
JavaSerializer		 - When running on a standalone deploy cluster or a - Mesos cluster in "coarse-grained" - sharing mode, the maximum amount of CPU cores to request for the application from - across the cluster (not from each machine). If not set, the default will be - spark.deploy.defaultCores on Spark's standalone cluster manager, or - infinite (all available cores) on Mesos. + Class to use for serializing objects that will be sent over the network or need to be cached + in serialized form. The default of Java serialization works with any Serializable Java object but is + quite slow, so we recommend using org.apache.spark.serializer.KryoSerializer + and configuring Kryo serialization when speed is necessary. Can be any subclass of + org.apache.spark.Serializer.
+ + +Apart from these, the following properties are also available, and may be useful in some situations: + + + - - + + - - + + - - + + - - + + - - + + - - + + - - + + - - + + - - + + -
Property NameDefaultMeaning
spark.core.connection.handler.threads.min20spark.akka.askTimeout30 - The number of threads to keep in the thread pool handling acceptance of the new connection, even they are - idle. + This parameter defines the timeout to use for Akka ask operations. (in seconds)
spark.core.connection.handler.threads.max60spark.akka.lookupTimeout30 - The maximum number of threads to allow in the pool handling acceptance of the new connection. + This parameter defines the timeout to use for Akka lookup operations. (in seconds)
spark.core.connection.handler.threads.keepalive60spark.akka.batchSize15 - When the number of threads in the pool handling acceptance of the new connection is greater than the core, - this is the maximum time that excess idle threads will wait for new tasks before terminating. + The parameter specifies defines the maximum number of messages to be processed per actor + before the thread jumps to the next actor. Set to 1 for as fair as possible.
spark.core.connection.io.threads.min20spark.akka.frameSize10 - The number of threads to keep in the thread pool handling BufferMessage receive/send, even they are - idle. + Maximum message size to allow in "control plane" communication (for serialized tasks and task + results), in MB. Increase this if your tasks need to send back large results to the driver + (e.g. using collect() on a large dataset).
spark.core.connection.io.threads.max60spark.akka.logAkkaConfigfalse - The maximum number of threads to allow in the pool handling BufferMessage receive/send. + This parameter controls three logging behaviours of Akka: 1) Logging the events including: DisassociatedEvent, + AssociatedEvent, AssociationErrorEvent, RemotingListenEvent, RemotingShutdownEvent, RemotingErrorEvent; 2) Logging + the dead letters during running; 3) Logging the dead letters during the shutdown.
spark.core.connection.io.threads.keepalive60spark.akka.failure-detector.threshold300.0 - When the number of threads in the pool handling BufferMessage receive/send is greater than the core, - this is the maximum time that excess idle threads will wait for new tasks before terminating. + This is set to a larger value to disable failure detector that comes inbuilt akka. It can be enabled again, if you plan to use this feature (Not recommended). This maps to akka's `akka.remote.transport-failure-detector.threshold`. Tune this in combination of `spark.akka.heartbeat.pauses` and `spark.akka.heartbeat.interval` if you need to.
spark.core.connection.connect.threads.min20spark.akka.heartbeat.interval1000 - The number of threads to keep in the thread pool handling the establishment of the connection, even they are - idle. + This is set to a larger value to disable failure detector that comes inbuilt akka. It can be enabled again, if you plan to use this feature (Not recommended). A larger interval value in seconds reduces network overhead and a smaller value ( ~ 1 s) might be more informative for akka's failure detector. Tune this in combination of `spark.akka.heartbeat.pauses` and `spark.akka.failure-detector.threshold` if you need to. Only positive use case for using failure detector can be, a sensistive failure detector can help evict rogue executors really quick. However this is usually not the case as gc pauses and network lags are expected in a real spark cluster. Apart from that enabling this leads to a lot of exchanges of heart beats between nodes leading to flooding the network with those.
spark.core.connection.connect.threads.max60spark.akka.heartbeat.pauses600 - The maximum number of threads to allow in the pool handling the establishment of the connection. + This is set to a larger value to disable failure detector that comes inbuilt akka. It can be enabled again, if you plan to use this feature (Not recommended). Acceptable heart beat pause in seconds for akka. This can be used to control sensitivity to gc pauses. Tune this in combination of `spark.akka.heartbeat.interval` and `spark.akka.failure-detector.threshold` if you need to.
spark.core.connection.connect.threads.keepalive60spark.akka.timeout100 - When the number of threads in the pool handling the establishment of the connection is greater than the core, - this is the maximum time that excess idle threads will wait for new tasks before terminating. + Communication timeout between Spark nodes, in seconds.
- - -Apart from these, the following properties are also available, and may be useful in some situations: - - - - - + + - - + + - - + + - - + + - - + + - - + + - - + + - - + + - - + + - - + + - - + + - - + + - - + + - - + + - - + + - - + + - - + + - - + + - - + + - - + + - - + + - - + + - - + + - - + + - - + + + + + + + + + + + + @@ -383,33 +400,6 @@ Apart from these, the following properties are also available, and may be useful exceeded" exception inside Kryo. Note that there will be one buffer per core on each worker. - - - - - - - - - - - - - - - @@ -418,168 +408,130 @@ Apart from these, the following properties are also available, and may be useful node locality and search immediately for rack locality (if your cluster has rack information). - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - + + + - - + + - - + + - - + + - - + + - - + + - - + + - - + + - - + + - - + + - - + + - - + + - - + + - - + + - - + + - @@ -595,6 +547,17 @@ Apart from these, the following properties are also available, and may be useful reduce the number of disk seeks and system calls made in creating intermediate shuffle files. + + + + + @@ -602,6 +565,14 @@ Apart from these, the following properties are also available, and may be useful This parameter sets the timeout threshold for the Netty connection when shuffling. (in milliseconds) + + + + + @@ -610,6 +581,27 @@ Apart from these, the following properties are also available, and may be useful threshold is specified by spark.shuffle.memoryFraction. + + + + + + + + + + + + + + + @@ -624,6 +616,13 @@ Apart from these, the following properties are also available, and may be useful How often Spark will check for tasks to speculate, in milliseconds. + + + + + @@ -632,89 +631,89 @@ Apart from these, the following properties are also available, and may be useful - - + + - - + + - - + + - - + + - - + + - - + + - - + + - - + + - - + + - - + + + + + + +
Property NameDefaultMeaning
spark.default.parallelism8spark.akka.threads4 - Default number of tasks to use across the cluster for distributed shuffle operations (groupByKey, - reduceByKey, etc) when not set by user. + Number of actor threads to use for communication. Can be useful to increase on large clusters + when the driver has a lot of CPU cores.
spark.storage.blockManagerTimeoutIntervalMs60000spark.broadcast.blockSize4096 - This parameter controls two values: First, the interval length for the BlockManagers running on the executors to send hearbeat - to the BlockManager running on the driver. Please note that in Spark the actual frequency is the value set with this parameter divided by 4. - Second, this value also controls the default timeout threshold of the slave node, which is spark.storage.blockManagerTimeoutIntervalMs * 3. - (In milliseconds) + Size of each piece of a block in kilobytes for TorrentBroadcastFactory. + Too large a value decreases parallelism during broadcast (makes it slower); however, if it is too small, BlockManager might take a performance hit.
spark.storage.blockManagerSlaveTimeoutMsspark.storage.blockManagerTimeoutIntervalMs * 3spark.broadcast.compresstrue - Number of milliseconds after which the BlockManagerMasterActor considers a BlockManager running on the executor lost if it receives no heartbeats. + Whether to compress broadcast variables before sending them. Generally a good idea.
spark.storage.memoryFraction0.6spark.broadcast.factoryorg.apache.spark.broadcast.
HttpBroadcastFactory		 - Fraction of Java heap to use for Spark's memory cache. This should not be larger than the "old" - generation of objects in the JVM, which by default is given 0.6 of the heap, but you can increase - it if you configure your own old generation size. + Which broadcast implementation to use.
spark.shuffle.memoryFraction0.3spark.cleaner.ttl(infinite) - Fraction of Java heap to use for aggregation and cogroups during shuffles, if - spark.shuffle.spill is true. At any given time, the collective size of - all in-memory maps used for shuffles is bounded by this limit, beyond which the contents will - begin to spill to disk. If spills are often, consider increasing this value at the expense of - spark.storage.memoryFraction. + Duration (seconds) of how long Spark will remember any metadata (stages generated, tasks generated, etc.). + Periodic cleanups will ensure that metadata older than this duration will be forgetten. This is + useful for running Spark for many hours / days (for example, running 24/7 in case of Spark Streaming + applications). Note that any RDD that persists in memory for more than this duration will be cleared as well.
spark.mesos.coarsefalsespark.closure.serializerorg.apache.spark.serializer.
JavaSerializer		 - If set to "true", runs over Mesos clusters in - "coarse-grained" sharing mode, - where Spark acquires one long-lived Mesos task on each machine instead of one Mesos task per Spark task. - This gives lower-latency scheduling for short queries, but leaves resources in use for the whole - duration of the Spark job. + Serializer class to use for closures. Generally Java is fine unless your distributed functions + (e.g. map functions) reference large objects in the driver program.
spark.mesos.extra.cores0spark.core.connection.handler.threads.min20 - When using Mesos, the user can specify how many cores each task can use in addition to the cores it gets from - the resource offer. + The number of threads to keep in the thread pool handling acceptance of the new connection, even they are + idle.
spark.ui.port4040spark.core.connection.handler.threads.max60 - Port for your application's dashboard, which shows memory and workload data + The maximum number of threads to allow in the pool handling acceptance of the new connection.
spark.ui.retainedStages1000spark.core.connection.handler.threads.keepalive60 - How many stages the Spark UI remembers before garbage collecting. + When the number of threads in the pool handling acceptance of the new connection is greater than the core, + this is the maximum time that excess idle threads will wait for new tasks before terminating.
spark.shuffle.compresstruespark.core.connection.io.threads.min20 - Whether to compress map output files. Generally a good idea. + The number of threads to keep in the thread pool handling BufferMessage receive/send, even they are + idle.
spark.shuffle.copier.threads6spark.core.connection.io.threads.max60 - When using Netty, this parameter specifies the number of concurrent threads fetching the shuffle results. This parameter - is only valid when spark.shuffle.use.netty is set to true. + The maximum number of threads to allow in the pool handling BufferMessage receive/send.
spark.shuffle.use.nettyfalsespark.core.connection.io.threads.keepalive60 - If use Netty to fetch map output in shuffle. + When the number of threads in the pool handling BufferMessage receive/send is greater than the core, + this is the maximum time that excess idle threads will wait for new tasks before terminating.
spark.shuffle.sender.port0spark.core.connection.connect.threads.min20 - When using Netty to send result for shuffle, this parameter specifies the port number for data transfer. - This parameter is only valid when spark.shuffle.use.netty is set to true. + The number of threads to keep in the thread pool handling the establishment of the connection, even they are + idle.
spark.shuffle.syncfalsespark.core.connection.connect.threads.max60 - If this parameter is set to true, Spark will force all outstanding writing for the Map outputs be flushed to the disk when closing the writers. + The maximum number of threads to allow in the pool handling the establishment of the connection.
spark.shuffle.spill.compresstruespark.core.connection.connect.threads.keepalive60 - Whether to compress data spilled during shuffles. + When the number of threads in the pool handling the establishment of the connection is greater than the core, + this is the maximum time that excess idle threads will wait for new tasks before terminating.
spark.broadcast.compresstruespark.dead.worker.persistence15 - Whether to compress broadcast variables before sending them. Generally a good idea. + This parameter controls when Spark will remove the worker information from the UI if it is dead. + If it has been longer than (spark.dead.worker.persistence + 1) * spark.worker.timeout + seconds since the worker is dead, Spark will remove it from UI.
spark.rdd.compressfalsespark.default.parallelism8 - Whether to compress serialized RDD partitions (e.g. for StorageLevel.MEMORY_ONLY_SER). - Can save substantial space at the cost of some extra CPU time. + Default number of tasks to use across the cluster for distributed shuffle operations (groupByKey, + reduceByKey, etc) when not set by user.
spark.io.compression.codecorg.apache.spark.io.
LZFCompressionCodec		spark.deploy.defaultCores(infinite) - The codec used to compress internal data such as RDD partitions and shuffle outputs. By default, Spark provides two - codecs: org.apache.spark.io.LZFCompressionCodec and org.apache.spark.io.SnappyCompressionCodec. + Default number of cores to give to applications in Spark's standalone mode if they don't + set spark.cores.max. If not set, applications always get all available + cores unless they configure spark.cores.max themselves. + Set this lower on a shared cluster to prevent users from grabbing + the whole cluster by default.
+ Note: this setting needs to be configured in the standalone cluster master, not in individual + applications; you can set it through SPARK_JAVA_OPTS in spark-env.sh.
spark.io.compression.snappy.block.size32768spark.deploy.retainedApplications200 - Block size (in bytes) used in Snappy compression, in the case when Snappy compression codec is used. + The maximum number of records on completed applications kept in the Master node. When the completed + application number exceeds the threshold, the first Max(spark.deploy.retainedApplications, 1) will be + discarded.
spark.scheduler.modeFIFOspark.deploy.spreadOuttrue - The scheduling mode between - jobs submitted to the same SparkContext. Can be set to FAIR - to use fair sharing instead of queueing jobs one after another. Useful for - multi-user services. + Whether the standalone cluster manager should spread applications out across nodes or try + to consolidate them onto as few nodes as possible. Spreading out is usually better for + data locality in HDFS, but consolidating is more efficient for compute-intensive workloads.
+ Note: this setting needs to be configured in the standalone cluster master, not in individual + applications; you can set it through SPARK_JAVA_OPTS in spark-env.sh.
spark.scheduler.revive.interval1000spark.diskStore.subDirectories64 - The interval length for the scheduler to revive the worker resource offers to run tasks. (in milliseconds) + Specifies how many subdirectories under each local directory. Setting more sub directories under each local directory + can help to avoid the fat inode.
spark.reducer.maxMbInFlight48spark.driver.host(local hostname) - Maximum size (in megabytes) of map outputs to fetch simultaneously from each reduce task. Since - each output requires us to create a buffer to receive it, this represents a fixed memory overhead - per reduce task, so keep it small unless you have a large amount of memory. + Hostname or IP address for the driver to listen on.
spark.resultGetter.threads4spark.driver.port(random) - Set the size of thread pool in TaskScheduler to deserialize and get the task results from remote end when necessary. - The size of the thread pool is fixed. + Port for the driver to listen on.
spark.repl.class.urinullspark.files.overwritefalse - This parameter specifies the URI of the class defined by REPL as user-defined code. + Whether to overwrite files added through SparkContext.addFile() when the target file exists and its contents do not match those of the source.
spark.closure.serializerorg.apache.spark.serializer.
JavaSerializer		spark.io.compression.codecorg.apache.spark.io.
LZFCompressionCodec		 - Serializer class to use for closures. Generally Java is fine unless your distributed functions - (e.g. map functions) reference large objects in the driver program. + The codec used to compress internal data such as RDD partitions and shuffle outputs. By default, Spark provides two + codecs: org.apache.spark.io.LZFCompressionCodec and org.apache.spark.io.SnappyCompressionCodec. +
spark.io.compression.snappy.block.size32768 + Block size (in bytes) used in Snappy compression, in the case when Snappy compression codec is used. +
spark.jars(not set) + The external jars to be propagated to the executors from the driver. Users can specify multiple jars by separating them + with ",".
spark.broadcast.factoryorg.apache.spark.broadcast.
HttpBroadcastFactory		 - Which broadcast implementation to use. -
spark.locality.wait3000 - Number of milliseconds to wait to launch a data-local task before giving up and launching it - on a less-local node. The same wait will be used to step through multiple locality levels - (process-local, node-local, rack-local and then any). It is also possible to customize the - waiting time for each level by setting spark.locality.wait.node, etc. - You should increase this setting if your tasks are long and see poor locality, but the - default usually works well. -
spark.locality.wait.processspark.locality.wait - Customize the locality wait for process locality. This affects tasks that attempt to access - cached data in a particular executor process. -
spark.locality.wait.node spark.locality.wait
spark.locality.wait.rackspark.locality.wait - Customize the locality wait for rack locality. -
spark.worker.timeout60 - Number of seconds after which the standalone deploy master considers a worker lost if it - receives no heartbeats. -
spark.akka.askTimeout30 - This parameter defines the timeout to use for Akka ask operations. (in seconds) -
spark.akka.lookupTimeout30 - This parameter defines the timeout to use for Akka lookup operations. (in seconds) -
spark.akka.batchSize15 - The parameter specifies defines the maximum number of messages to be processed per actor - before the thread jumps to the next actor. Set to 1 for as fair as possible. -
spark.akka.frameSize10 - Maximum message size to allow in "control plane" communication (for serialized tasks and task - results), in MB. Increase this if your tasks need to send back large results to the driver - (e.g. using collect() on a large dataset). -
spark.akka.logAkkaConfigfalse
spark.locality.wait.processspark.locality.wait - This parameter controls three logging behaviours of Akka: 1) Logging the events including: DisassociatedEvent, - AssociatedEvent, AssociationErrorEvent, RemotingListenEvent, RemotingShutdownEvent, RemotingErrorEvent; 2) Logging - the dead letters during running; 3) Logging the dead letters during the shutdown. + Customize the locality wait for process locality. This affects tasks that attempt to access + cached data in a particular executor process.
spark.akka.threads4spark.locality.wait.rackspark.locality.wait - Number of actor threads to use for communication. Can be useful to increase on large clusters - when the driver has a lot of CPU cores. + Customize the locality wait for rack locality.
spark.akka.timeout100spark.locality.wait3000 - Communication timeout between Spark nodes, in seconds. + Number of milliseconds to wait to launch a data-local task before giving up and launching it + on a less-local node. The same wait will be used to step through multiple locality levels + (process-local, node-local, rack-local and then any). It is also possible to customize the + waiting time for each level by setting spark.locality.wait.node, etc. + You should increase this setting if your tasks are long and see poor locality, but the + default usually works well.
spark.akka.heartbeat.pauses600spark.logging.exceptionPrintInterval10000 - This is set to a larger value to disable failure detector that comes inbuilt akka. It can be enabled again, if you plan to use this feature (Not recommended). Acceptable heart beat pause in seconds for akka. This can be used to control sensitivity to gc pauses. Tune this in combination of `spark.akka.heartbeat.interval` and `spark.akka.failure-detector.threshold` if you need to. + How frequently the driver reprints duplicate exceptions in full. (in milliseconds)
spark.akka.failure-detector.threshold300.0spark.logConffalse - This is set to a larger value to disable failure detector that comes inbuilt akka. It can be enabled again, if you plan to use this feature (Not recommended). This maps to akka's `akka.remote.transport-failure-detector.threshold`. Tune this in combination of `spark.akka.heartbeat.pauses` and `spark.akka.heartbeat.interval` if you need to. + Log the supplied SparkConf as INFO at start of spark context.
spark.akka.heartbeat.interval1000spark.mesos.coarsefalse - This is set to a larger value to disable failure detector that comes inbuilt akka. It can be enabled again, if you plan to use this feature (Not recommended). A larger interval value in seconds reduces network overhead and a smaller value ( ~ 1 s) might be more informative for akka's failure detector. Tune this in combination of `spark.akka.heartbeat.pauses` and `spark.akka.failure-detector.threshold` if you need to. Only positive use case for using failure detector can be, a sensistive failure detector can help evict rogue executors really quick. However this is usually not the case as gc pauses and network lags are expected in a real spark cluster. Apart from that enabling this leads to a lot of exchanges of heart beats between nodes leading to flooding the network with those. + If set to "true", runs over Mesos clusters in + "coarse-grained" sharing mode, + where Spark acquires one long-lived Mesos task on each machine instead of one Mesos task per Spark task. + This gives lower-latency scheduling for short queries, but leaves resources in use for the whole + duration of the Spark job.
spark.driver.host(local hostname)spark.mesos.extra.cores0 - Hostname or IP address for the driver to listen on. + When using Mesos, the user can specify how many cores each task can use in addition to the cores it gets from + the resource offer.
spark.driver.port(random)spark.rdd.compressfalse - Port for the driver to listen on. + Whether to compress serialized RDD partitions (e.g. for StorageLevel.MEMORY_ONLY_SER). + Can save substantial space at the cost of some extra CPU time.
spark.cleaner.ttl(infinite)spark.reducer.maxMbInFlight48 - Duration (seconds) of how long Spark will remember any metadata (stages generated, tasks generated, etc.). - Periodic cleanups will ensure that metadata older than this duration will be forgetten. This is - useful for running Spark for many hours / days (for example, running 24/7 in case of Spark Streaming - applications). Note that any RDD that persists in memory for more than this duration will be cleared as well. + Maximum size (in megabytes) of map outputs to fetch simultaneously from each reduce task. Since + each output requires us to create a buffer to receive it, this represents a fixed memory overhead + per reduce task, so keep it small unless you have a large amount of memory.
spark.starvation.timeout15000spark.repl.class.urinull - Threshold above which we warn user initial TaskSet may be starved. (in milliseconds) + This parameter specifies the URI of the class defined by REPL as user-defined code.
spark.streaming.blockInterval200spark.resultGetter.threads4 - Duration (milliseconds) of how long to batch new objects coming from network receivers used - in Spark Streaming. + Set the size of thread pool in TaskScheduler to deserialize and get the task results from remote end when necessary. + The size of the thread pool is fixed.
spark.streaming.unpersistfalsespark.scheduler.modeFIFO - Force RDDs generated and persisted by Spark Streaming to be automatically unpersisted from - Spark's memory. Setting this to true is likely to reduce Spark's RDD memory usage. + The scheduling mode between + jobs submitted to the same SparkContext. Can be set to FAIR + to use fair sharing instead of queueing jobs one after another. Useful for + multi-user services.
spark.task.maxFailures4spark.scheduler.revive.interval1000 - Number of individual task failures before giving up on the job. - Should be greater than or equal to 1. Number of allowed retries = this value - 1. + The interval length for the scheduler to revive the worker resource offers to run tasks. (in milliseconds)
spark.task.cpus1spark.shuffle.compresstrue - CPUs to request per task + Whether to compress map output files. Generally a good idea.
spark.broadcast.blockSize4096spark.shuffle.copier.threads6 - Size of each piece of a block in kilobytes for TorrentBroadcastFactory. - Too large a value decreases parallelism during broadcast (makes it slower); however, if it is too small, BlockManager might take a performance hit. + When using Netty, this parameter specifies the number of concurrent threads fetching the shuffle results. This parameter + is only valid when spark.shuffle.use.netty is set to true.
spark.shuffle.consolidateFiles false
spark.shuffle.memoryFraction0.3 + Fraction of Java heap to use for aggregation and cogroups during shuffles, if + spark.shuffle.spill is true. At any given time, the collective size of + all in-memory maps used for shuffles is bounded by this limit, beyond which the contents will + begin to spill to disk. If spills are often, consider increasing this value at the expense of + spark.storage.memoryFraction. +
spark.shuffle.netty.connect.timeout 60000
spark.shuffle.sender.port0 + When using Netty to send result for shuffle, this parameter specifies the port number for data transfer. + This parameter is only valid when spark.shuffle.use.netty is set to true. +
spark.shuffle.spill true
spark.shuffle.spill.compresstrue + Whether to compress data spilled during shuffles. +
spark.shuffle.syncfalse + If this parameter is set to true, Spark will force all outstanding writing for the Map outputs be flushed to the disk when closing the writers. +
spark.shuffle.use.nettyfalse + If use Netty to fetch map output in shuffle. +
spark.speculation false
spark.speculation.multiplier1.5 + How many times slower a task is than the median to be considered for speculation. +
spark.speculation.quantile 0.75
spark.speculation.multiplier1.5spark.starvation.timeout15000 - How many times slower a task is than the median to be considered for speculation. + Threshold above which we warn user initial TaskSet may be starved. (in milliseconds)
spark.jars(not set)spark.storage.blockManagerSlaveTimeoutMsspark.storage.blockManagerTimeoutIntervalMs * 3 - The external jars to be propagated to the executors from the driver. Users can specify multiple jars by separating them - with ",". + Number of milliseconds after which the BlockManagerMasterActor considers a BlockManager running on the executor lost if it receives no heartbeats.
spark.logConffalsespark.storage.blockManagerTimeoutIntervalMs60000 - Log the supplied SparkConf as INFO at start of spark context. + This parameter controls two values: First, the interval length for the BlockManagers running on the executors to send hearbeat + to the BlockManager running on the driver. Please note that in Spark the actual frequency is the value set with this parameter divided by 4. + Second, this value also controls the default timeout threshold of the slave node, which is spark.storage.blockManagerTimeoutIntervalMs * 3. + (In milliseconds)
spark.logging.exceptionPrintInterval10000spark.storage.memoryFraction0.6 - How frequently the driver reprints duplicate exceptions in full. (in milliseconds) + Fraction of Java heap to use for Spark's memory cache. This should not be larger than the "old" + generation of objects in the JVM, which by default is given 0.6 of the heap, but you can increase + it if you configure your own old generation size.
spark.dead.worker.persistence15spark.streaming.blockInterval200 - This parameter controls when Spark will remove the worker information from the UI if it is dead. - If it has been longer than (spark.dead.worker.persistence + 1) * spark.worker.timeout - seconds since the worker is dead, Spark will remove it from UI. + Duration (milliseconds) of how long to batch new objects coming from network receivers used + in Spark Streaming.
spark.deploy.defaultCores(infinite)spark.streaming.unpersistfalse - Default number of cores to give to applications in Spark's standalone mode if they don't - set spark.cores.max. If not set, applications always get all available - cores unless they configure spark.cores.max themselves. - Set this lower on a shared cluster to prevent users from grabbing - the whole cluster by default.
- Note: this setting needs to be configured in the standalone cluster master, not in individual - applications; you can set it through SPARK_JAVA_OPTS in spark-env.sh. + Force RDDs generated and persisted by Spark Streaming to be automatically unpersisted from + Spark's memory. Setting this to true is likely to reduce Spark's RDD memory usage.
spark.deploy.retainedApplications200spark.task.cpus1 - The maximum number of records on completed applications kept in the Master node. When the completed - application number exceeds the threshold, the first Max(spark.deploy.retainedApplications, 1) will be - discarded. + CPUs to request per task
spark.deploy.spreadOuttruespark.task.maxFailures4 - Whether the standalone cluster manager should spread applications out across nodes or try - to consolidate them onto as few nodes as possible. Spreading out is usually better for - data locality in HDFS, but consolidating is more efficient for compute-intensive workloads.
- Note: this setting needs to be configured in the standalone cluster master, not in individual - applications; you can set it through SPARK_JAVA_OPTS in spark-env.sh. + Number of individual task failures before giving up on the job. + Should be greater than or equal to 1. Number of allowed retries = this value - 1.
spark.diskStore.subDirectories64spark.ui.port4040 - Specifies how many subdirectories under each local directory. Setting more sub directories under each local directory - can help to avoid the fat inode. + Port for your application's dashboard, which shows memory and workload data
spark.files.overwritefalsespark.ui.retainedStages1000 - Whether to overwrite files added through SparkContext.addFile() when the target file exists and its contents do not match those of the source. + How many stages the Spark UI remembers before garbage collecting. +
spark.worker.timeout60 + Number of seconds after which the standalone deploy master considers a worker lost if it + receives no heartbeats.
From c4d881700bfd6b8fa4b73b096ddf995e9358717c Mon Sep 17 00:00:00 2001 From: Nan Zhu Date: Thu, 6 Mar 2014 19:48:58 -0500 Subject: [PATCH 28/33] style fix --- .../main/scala/org/apache/spark/repl/ExecutorClassLoader.scala | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/repl/src/main/scala/org/apache/spark/repl/ExecutorClassLoader.scala b/repl/src/main/scala/org/apache/spark/repl/ExecutorClassLoader.scala index cdfe8abdd6f44..87a0333f24b15 100644 --- a/repl/src/main/scala/org/apache/spark/repl/ExecutorClassLoader.scala +++ b/repl/src/main/scala/org/apache/spark/repl/ExecutorClassLoader.scala @@ -32,8 +32,7 @@ import org.objectweb.asm.Opcodes._ * A ClassLoader that reads classes from a Hadoop FileSystem or HTTP URI, * used to load classes defined by the interpreter when the REPL is used */ -class ExecutorClassLoader(classUri: String, parent: ClassLoader) - extends ClassLoader(parent) { +class ExecutorClassLoader(classUri: String, parent: ClassLoader) extends ClassLoader(parent) { val uri = new URI(classUri) val directory = uri.getPath From 7029d72cc85356a57285c3096a0f61fb96a67a6a Mon Sep 17 00:00:00 2001 From: CodingCat Date: Thu, 6 Mar 2014 23:38:22 -0500 Subject: [PATCH 29/33] sync with 33baf14b04 --- docs/configuration.md | 52 +++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 52 insertions(+) diff --git a/docs/configuration.md b/docs/configuration.md index 22bcedd9e4a07..5bbdeec42ebeb 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -170,6 +170,22 @@ Apart from these, the following properties are also available, and may be useful when the driver has a lot of CPU cores. + + spark.authenticate + false + + Whether spark authenticates its internal connections. See spark.authenticate.secret if not + running on Yarn. + + + + spark.authenticate.secret + None + + Set 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.broadcast.blockSize 4096 @@ -210,6 +226,14 @@ Apart from these, the following properties are also available, and may be useful (e.g. map functions) reference large objects in the driver program. + + spark.core.connection.auth.wait.timeout + 30 + + Number of seconds for the connection to wait for authentication to occur before timing + out and giving up. + + spark.core.connection.handler.threads.min 20 @@ -694,6 +718,26 @@ Apart from these, the following properties are also available, and may be useful Should be greater than or equal to 1. Number of allowed retries = this value - 1. + + spark.ui.acls.enable + false + + Whether spark web ui acls should are enabled. If enabled, this checks to see if the user has + access permissions to view the web ui. See spark.ui.view.acls for more details. + Also note this requires the user to be known, if the user comes across as null no checks + are done. Filters can be used to authenticate and set the user. + + + + spark.ui.filters + None + + Comma 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' + (e.g.-Dspark.ui.filters=com.test.filter1 -Dspark.com.test.filter1.params='param1=foo,param2=testing') + + spark.ui.port 4040 @@ -708,6 +752,14 @@ Apart from these, the following properties are also available, and may be useful How many stages the Spark UI remembers before garbage collecting. + + spark.ui.view.acls + Empty + + Comma 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. + + spark.worker.timeout 60 From c698465fe7f15008abfa11e2395a5cbe82eee59a Mon Sep 17 00:00:00 2001 From: CodingCat Date: Thu, 6 Mar 2014 23:41:30 -0500 Subject: [PATCH 30/33] sync with 40566e10aa --- docs/configuration.md | 11 +++++++++++ 1 file changed, 11 insertions(+) diff --git a/docs/configuration.md b/docs/configuration.md index 5bbdeec42ebeb..5abfe31e2de43 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -541,6 +541,17 @@ Apart from these, the following properties are also available, and may be useful The interval length for the scheduler to revive the worker resource offers to run tasks. (in milliseconds) + + spark.serializer.objectStreamReset + 10000 + + When serializing using org.apache.spark.serializer.JavaSerializer, the serializer caches + objects to prevent writing redundant data, however that stops garbage collection of those + objects. By calling 'reset' you flush that info from the serializer, and allow old + objects to be collected. To turn off this periodic reset set it to a value of <= 0. + By default it will reset the serializer every 10,000 objects. + + spark.shuffle.compress true From a1ecf0b159c498bfcb97e7fb3304058a971ee186 Mon Sep 17 00:00:00 2001 From: CodingCat Date: Mon, 10 Mar 2014 13:48:29 -0400 Subject: [PATCH 31/33] sync with f6f9d02 --- docs/configuration.md | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/docs/configuration.md b/docs/configuration.md index 5abfe31e2de43..d604d0a2e4b73 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -375,6 +375,14 @@ Apart from these, the following properties are also available, and may be useful Port for the driver to listen on. + + spark.files.fetchTimeout + false + + Communication timeout to use when fetching files added through SparkContext.addFile() from + the driver. + + spark.files.overwrite false From 54ebdf16034adcda5e2d4d0b2ec8675f598f2a80 Mon Sep 17 00:00:00 2001 From: CodingCat Date: Mon, 31 Mar 2014 11:58:01 -0400 Subject: [PATCH 32/33] sync with latest PRs --- docs/configuration.md | 18 ++++++++++++++++-- 1 file changed, 16 insertions(+), 2 deletions(-) diff --git a/docs/configuration.md b/docs/configuration.md index d604d0a2e4b73..816c6be78217e 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -50,6 +50,20 @@ there are at least five properties that you will commonly want to control: infinite (all available cores) on Mesos. + + spark.eventLog.compress + false + + Whether to compress logged events, if spark.eventLog.enabled is true. + + + + spark.eventLog.enabled + false + + Whether to log spark events, useful for reconstructing the Web UI after the application has finished. + + spark.executor.memory 512m @@ -478,7 +492,7 @@ Apart from these, the following properties are also available, and may be useful spark.logConf false - Log the supplied SparkConf as INFO at start of spark context. + Whether to log the supplied SparkConf as INFO at start of spark context. @@ -726,7 +740,7 @@ Apart from these, the following properties are also available, and may be useful spark.task.cpus 1 - CPUs to request per task + Number of cores to allocate for each task. From 273e5d8467f814656fe7198c94b89ca0a7b5946a Mon Sep 17 00:00:00 2001 From: CodingCat Date: Thu, 17 Apr 2014 07:33:45 -0400 Subject: [PATCH 33/33] recover the scala file (moving to a separate PR) --- core/src/main/scala/org/apache/spark/util/AkkaUtils.scala | 2 +- .../main/scala/org/apache/spark/repl/ExecutorClassLoader.scala | 3 ++- 2 files changed, 3 insertions(+), 2 deletions(-) diff --git a/core/src/main/scala/org/apache/spark/util/AkkaUtils.scala b/core/src/main/scala/org/apache/spark/util/AkkaUtils.scala index 6c9f266b455b3..f26ed47e58046 100644 --- a/core/src/main/scala/org/apache/spark/util/AkkaUtils.scala +++ b/core/src/main/scala/org/apache/spark/util/AkkaUtils.scala @@ -108,6 +108,6 @@ private[spark] object AkkaUtils { /** Returns the default Spark timeout to use for Akka remote actor lookup. */ def lookupTimeout(conf: SparkConf): FiniteDuration = { - Duration.create(conf.getLong("spark.akka.lookupTimeout", 30), "seconds") + Duration.create(conf.get("spark.akka.lookupTimeout", "30").toLong, "seconds") } } diff --git a/repl/src/main/scala/org/apache/spark/repl/ExecutorClassLoader.scala b/repl/src/main/scala/org/apache/spark/repl/ExecutorClassLoader.scala index 87a0333f24b15..e3bcf7f30ac8d 100644 --- a/repl/src/main/scala/org/apache/spark/repl/ExecutorClassLoader.scala +++ b/repl/src/main/scala/org/apache/spark/repl/ExecutorClassLoader.scala @@ -32,7 +32,8 @@ import org.objectweb.asm.Opcodes._ * A ClassLoader that reads classes from a Hadoop FileSystem or HTTP URI, * used to load classes defined by the interpreter when the REPL is used */ -class ExecutorClassLoader(classUri: String, parent: ClassLoader) extends ClassLoader(parent) { +class ExecutorClassLoader(classUri: String, parent: ClassLoader) +extends ClassLoader(parent) { val uri = new URI(classUri) val directory = uri.getPath