Moved ExternalShuffleService to the deploy package + other minor tweaks.

Author: dragos

core/src/main/scala/org/apache/spark/deploy/worker/ExternalShuffleService.scala renamed to core/src/main/scala/org/apache/spark/deploy/ExternalShuffleService.scala

@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
-package org.apache.spark.deploy.worker
+package org.apache.spark.deploy
 
 import org.apache.spark.{Logging, SparkConf, SecurityManager}
 import org.apache.spark.util.Utils
@@ -24,8 +24,8 @@ import org.apache.spark.network.netty.SparkTransportConf
 import org.apache.spark.network.sasl.SaslRpcHandler
 import org.apache.spark.network.server.TransportServer
 import org.apache.spark.network.shuffle.ExternalShuffleBlockHandler
-
 import java.util.concurrent.CountDownLatch
+import org.apache.spark.annotation.DeveloperApi
 
 /**
  * Provides a server from which Executors can read shuffle files (rather than reading directly from
@@ -34,7 +34,7 @@ import java.util.concurrent.CountDownLatch
  *
  * Optionally requires SASL authentication in order to read. See [[SecurityManager]].
  */
-private[worker]
+private[deploy]
 class ExternalShuffleService(sparkConf: SparkConf, securityManager: SecurityManager)
   extends Logging {
 
@@ -54,14 +54,19 @@ class ExternalShuffleService(sparkConf: SparkConf, securityManager: SecurityMana
   /** Starts the external shuffle service if the user has configured us to. */
   def startIfEnabled() {
     if (enabled) {
-      require(server == null, "Shuffle server already started")
-      logInfo(s"Starting shuffle service on port $port with useSasl = $useSasl")
-      server = transportContext.createServer(port)
+      start()
     }
   }
 
+  /** Start the external shuffle service */
+  def start() {
+    require(server == null, "Shuffle server already started")
+    logInfo(s"Starting shuffle service on port $port with useSasl = $useSasl")
+    server = transportContext.createServer(port)
+  }
+
   def stop() {
-    if (enabled && server != null) {
+    if (server != null) {
       server.close()
       server = null
     }
@@ -86,14 +91,15 @@ object ExternalShuffleService extends Logging {
     // and we assume the user really wants it to be running
     sparkConf.set("spark.shuffle.service.enabled", "true")
     server = new ExternalShuffleService(sparkConf, securityManager)
+    server.start()
 
     installShutdownHook()
 
     // keep running until the process is terminated
     barrier.await()
   }
 
-  private def installShutdownHook() = {
+  private def installShutdownHook(): Unit = {
     Runtime.getRuntime.addShutdownHook(new Thread("External Shuffle Service shutdown thread") {
       override def run() {
         logInfo("Shutting down shuffle service.")

core/src/main/scala/org/apache/spark/deploy/worker/Worker.scala

@@ -34,6 +34,7 @@ import akka.remote.{DisassociatedEvent, RemotingLifecycleEvent}
 import org.apache.spark.{Logging, SecurityManager, SparkConf}
 import org.apache.spark.deploy.{Command, ExecutorDescription, ExecutorState}
 import org.apache.spark.deploy.DeployMessages._
+import org.apache.spark.deploy.ExternalShuffleService
 import org.apache.spark.deploy.master.{DriverState, Master}
 import org.apache.spark.deploy.worker.ui.WorkerWebUI
 import org.apache.spark.metrics.MetricsSystem

docs/job-scheduling.md

@@ -99,11 +99,6 @@ then set `yarn.nodemanager.aux-services.spark_shuffle.class` to
 `spark.shuffle.service.*` [configurations](configuration.html).
 4. Restart all `NodeManager`s in your cluster.
 
-If you wish to use the external shuffle service in a Mesos deployment you need to launch the
-service on each Mesos slave that might run Spark jobs. A launcher script is provided in the `sbin/`
-directory: `sbin/start-shuffle-service.sh`. You can later stop the service by running
-`sbin/stop-shuffle-service.sh`. Additional configuration options can be passed in `SPARK_SHUFFLE_OPTS`.
-
 ### Resource Allocation Policy
 
 At a high level, Spark should relinquish executors when they are no longer used and acquire

sbin/start-shuffle-service.sh

@@ -30,4 +30,4 @@ sbin="`cd "$sbin"; pwd`"
 . "$sbin/spark-config.sh"
 . "$SPARK_PREFIX/bin/load-spark-env.sh"
 
-exec "$sbin"/spark-daemon.sh start org.apache.spark.deploy.worker.ExternalShuffleService 1
+exec "$sbin"/spark-daemon.sh start org.apache.spark.deploy.ExternalShuffleService 1

sbin/stop-shuffle-service.sh

@@ -22,4 +22,4 @@
 sbin="`dirname "$0"`"
 sbin="`cd "$sbin"; pwd`"
 
-"$sbin"/spark-daemon.sh stop org.apache.spark.deploy.worker.ExternalShuffleService 1
+"$sbin"/spark-daemon.sh stop org.apache.spark.deploy.ExternalShuffleService 1