diff options
20 files changed, 456 insertions, 346 deletions
diff --git a/core/src/main/scala/spark/Accumulators.scala b/core/src/main/scala/spark/Accumulators.scala index 094a95d70e..6018d251d1 100644 --- a/core/src/main/scala/spark/Accumulators.scala +++ b/core/src/main/scala/spark/Accumulators.scala @@ -5,10 +5,13 @@ import java.io._ import scala.collection.mutable.Map class Accumulator[T] ( - @transient initialValue: T, param: AccumulatorParam[T]) extends Serializable -{ + @transient initialValue: T, + param: AccumulatorParam[T] + ) extends Serializable { + val id = Accumulators.newId - @transient var value_ = initialValue // Current value on master + @transient + var value_ = initialValue // Current value on master val zero = param.zero(initialValue) // Zero value to be passed to workers var deserialized = false @@ -39,14 +42,16 @@ trait AccumulatorParam[T] extends Serializable { // TODO: The multi-thread support in accumulators is kind of lame; check // if there's a more intuitive way of doing it right -private object Accumulators -{ +private object Accumulators { // TODO: Use soft references? => need to make readObject work properly then val originals = Map[Long, Accumulator[_]]() val localAccums = Map[Thread, Map[Long, Accumulator[_]]]() var lastId: Long = 0 - def newId: Long = synchronized { lastId += 1; return lastId } + def newId: Long = synchronized { + lastId += 1 + return lastId + } def register(a: Accumulator[_], original: Boolean): Unit = synchronized { if (original) { @@ -65,8 +70,9 @@ private object Accumulators // Get the values of the local accumulators for the current thread (by ID) def values: Map[Long, Any] = synchronized { val ret = Map[Long, Any]() - for ((id, accum) <- localAccums.getOrElse(Thread.currentThread, Map())) + for ((id, accum) <- localAccums.getOrElse(Thread.currentThread, Map())) { ret(id) = accum.value + } return ret } diff --git a/core/src/main/scala/spark/Cache.scala b/core/src/main/scala/spark/Cache.scala index 89befae1a4..24ac88c14f 100644 --- a/core/src/main/scala/spark/Cache.scala +++ b/core/src/main/scala/spark/Cache.scala @@ -2,7 +2,6 @@ package spark import java.util.concurrent.atomic.AtomicLong - /** * An interface for caches in Spark, to allow for multiple implementations. * Caches are used to store both partitions of cached RDDs and broadcast @@ -29,7 +28,6 @@ abstract class Cache { def put(key: Any, value: Any): Unit } - /** * A key namespace in a Cache. */ diff --git a/core/src/main/scala/spark/ClosureCleaner.scala b/core/src/main/scala/spark/ClosureCleaner.scala index 40bcbb5a9d..62d2c4cb12 100644 --- a/core/src/main/scala/spark/ClosureCleaner.scala +++ b/core/src/main/scala/spark/ClosureCleaner.scala @@ -9,7 +9,6 @@ import org.objectweb.asm.{ClassReader, MethodVisitor, Type} import org.objectweb.asm.commons.EmptyVisitor import org.objectweb.asm.Opcodes._ - object ClosureCleaner extends Logging { // Get an ASM class reader for a given class from the JAR that loaded it private def getClassReader(cls: Class[_]): ClassReader = { @@ -154,7 +153,6 @@ object ClosureCleaner extends Logging { } } - class FieldAccessFinder(output: Map[Class[_], Set[String]]) extends EmptyVisitor { override def visitMethod(access: Int, name: String, desc: String, sig: String, exceptions: Array[String]): MethodVisitor = { diff --git a/core/src/main/scala/spark/DAGScheduler.scala b/core/src/main/scala/spark/DAGScheduler.scala index 2e6cff4e52..be6756aa48 100644 --- a/core/src/main/scala/spark/DAGScheduler.scala +++ b/core/src/main/scala/spark/DAGScheduler.scala @@ -4,18 +4,28 @@ import java.util.concurrent.LinkedBlockingQueue import java.util.concurrent.TimeUnit import scala.collection.mutable.{ArrayBuffer, HashMap, HashSet, Map} -// A task created by the DAG scheduler. Knows its stage ID and map ouput tracker generation. +/** + * A task created by the DAG scheduler. Knows its stage ID and map ouput tracker generation. + */ abstract class DAGTask[T](val stageId: Int) extends Task[T] { val gen = SparkEnv.get.mapOutputTracker.getGeneration override def generation: Option[Long] = Some(gen) } -// A completion event passed by the underlying task scheduler to the DAG scheduler -case class CompletionEvent(task: DAGTask[_], reason: TaskEndReason, result: Any, accumUpdates: Map[Long, Any]) +/** + * A completion event passed by the underlying task scheduler to the DAG scheduler + */ +case class CompletionEvent( + task: DAGTask[_], + reason: TaskEndReason, + result: Any, + accumUpdates: Map[Long, Any]) -// Various possible reasons why a DAG task ended. The underlying scheduler is supposed -// to retry tasks several times for "ephemeral" failures, and only report back failures -// that require some old stages to be resubmitted, such as shuffle map fetch failures. +/** + * Various possible reasons why a DAG task ended. The underlying scheduler is supposed to retry + * tasks several times for "ephemeral" failures, and only report back failures that require some + * old stages to be resubmitted, such as shuffle map fetch failures. + */ sealed trait TaskEndReason case object Success extends TaskEndReason case class FetchFailed(serverUri: String, shuffleId: Int, mapId: Int, reduceId: Int) extends TaskEndReason @@ -23,11 +33,10 @@ case class ExceptionFailure(exception: Throwable) extends TaskEndReason case class OtherFailure(message: String) extends TaskEndReason /** - * A Scheduler subclass that implements stage-oriented scheduling. It computes - * a DAG of stages for each job, keeps track of which RDDs and stage outputs - * are materialized, and computes a minimal schedule to run the job. Subclasses - * only need to implement the code to send a task to the cluster and to report - * fetch failures (the submitTasks method, and code to add CompletionEvents). + * A Scheduler subclass that implements stage-oriented scheduling. It computes a DAG of stages for + * each job, keeps track of which RDDs and stage outputs are materialized, and computes a minimal + * schedule to run the job. Subclasses only need to implement the code to send a task to the cluster + * and to report fetch failures (the submitTasks method, and code to add CompletionEvents). */ private trait DAGScheduler extends Scheduler with Logging { // Must be implemented by subclasses to start running a set of tasks @@ -39,16 +48,15 @@ private trait DAGScheduler extends Scheduler with Logging { completionEvents.put(CompletionEvent(dagTask, reason, result, accumUpdates)) } - // The time, in millis, to wait for fetch failure events to stop coming in after - // one is detected; this is a simplistic way to avoid resubmitting tasks in the - // non-fetchable map stage one by one as more failure events come in + // The time, in millis, to wait for fetch failure events to stop coming in after one is detected; + // this is a simplistic way to avoid resubmitting tasks in the non-fetchable map stage one by one + // as more failure events come in val RESUBMIT_TIMEOUT = 2000L - // The time, in millis, to wake up between polls of the completion queue - // in order to potentially resubmit failed stages + // The time, in millis, to wake up between polls of the completion queue in order to potentially + // resubmit failed stages val POLL_TIMEOUT = 500L - private val completionEvents = new LinkedBlockingQueue[CompletionEvent] var nextStageId = 0 @@ -110,10 +118,8 @@ private trait DAGScheduler extends Scheduler with Logging { cacheTracker.registerRDD(r.id, r.splits.size) for (dep <- r.dependencies) { dep match { - case shufDep: ShuffleDependency[_,_,_] => - parents += getShuffleMapStage(shufDep) - case _ => - visit(dep.rdd) + case shufDep: ShuffleDependency[_,_,_] => parents += getShuffleMapStage(shufDep) + case _ => visit(dep.rdd) } } } @@ -135,10 +141,10 @@ private trait DAGScheduler extends Scheduler with Logging { dep match { case shufDep: ShuffleDependency[_,_,_] => val stage = getShuffleMapStage(shufDep) - if (!stage.isAvailable) + if (!stage.isAvailable) { missing += stage - case narrowDep: NarrowDependency[_] => - visit(narrowDep.rdd) + } + case narrowDep: NarrowDependency[_] => visit(narrowDep.rdd) } } } @@ -149,10 +155,11 @@ private trait DAGScheduler extends Scheduler with Logging { missing.toList } - override def runJob[T, U](finalRdd: RDD[T], func: (TaskContext, Iterator[T]) => U, - partitions: Seq[Int], allowLocal: Boolean) - (implicit m: ClassManifest[U]) - : Array[U] = { + override def runJob[T, U](finalRdd: RDD[T], + func: (TaskContext, Iterator[T]) => U, + partitions: Seq[Int], + allowLocal: Boolean + )(implicit m: ClassManifest[U]) : Array[U] = { val outputParts = partitions.toArray val numOutputParts: Int = partitions.size val finalStage = newStage(finalRdd, None) @@ -189,8 +196,9 @@ private trait DAGScheduler extends Scheduler with Logging { submitMissingTasks(stage) running += stage } else { - for (parent <- missing) + for (parent <- missing) { submitStage(parent) + } waiting += stage } } diff --git a/core/src/main/scala/spark/Executor.scala b/core/src/main/scala/spark/Executor.scala index 15693fc95f..83d2df4f94 100644 --- a/core/src/main/scala/spark/Executor.scala +++ b/core/src/main/scala/spark/Executor.scala @@ -27,8 +27,9 @@ class Executor extends org.apache.mesos.Executor with Logging { override def init(d: ExecutorDriver, args: ExecutorArgs) { // Read spark.* system properties from executor arg val props = Utils.deserialize[Array[(String, String)]](args.getData.toByteArray) - for ((key, value) <- props) + for ((key, value) <- props) { System.setProperty(key, value) + } // Make sure an appropriate class loader is set for remote actors RemoteActor.classLoader = getClass.getClassLoader @@ -45,7 +46,7 @@ class Executor extends org.apache.mesos.Executor with Logging { // Start worker thread pool threadPool = new ThreadPoolExecutor( - 1, 128, 600, TimeUnit.SECONDS, new SynchronousQueue[Runnable]) + 1, 128, 600, TimeUnit.SECONDS, new SynchronousQueue[Runnable]) } override def launchTask(d: ExecutorDriver, task: TaskDescription) { @@ -58,9 +59,9 @@ class Executor extends org.apache.mesos.Executor with Logging { val tid = desc.getTaskId.getValue logInfo("Running task ID " + tid) d.sendStatusUpdate(TaskStatus.newBuilder() - .setTaskId(desc.getTaskId) - .setState(TaskState.TASK_RUNNING) - .build()) + .setTaskId(desc.getTaskId) + .setState(TaskState.TASK_RUNNING) + .build()) try { SparkEnv.set(env) Thread.currentThread.setContextClassLoader(classLoader) @@ -72,27 +73,27 @@ class Executor extends org.apache.mesos.Executor with Logging { val accumUpdates = Accumulators.values val result = new TaskResult(value, accumUpdates) d.sendStatusUpdate(TaskStatus.newBuilder() - .setTaskId(desc.getTaskId) - .setState(TaskState.TASK_FINISHED) - .setData(ByteString.copyFrom(Utils.serialize(result))) - .build()) + .setTaskId(desc.getTaskId) + .setState(TaskState.TASK_FINISHED) + .setData(ByteString.copyFrom(Utils.serialize(result))) + .build()) logInfo("Finished task ID " + tid) } catch { case ffe: FetchFailedException => { val reason = ffe.toTaskEndReason d.sendStatusUpdate(TaskStatus.newBuilder() - .setTaskId(desc.getTaskId) - .setState(TaskState.TASK_FAILED) - .setData(ByteString.copyFrom(Utils.serialize(reason))) - .build()) + .setTaskId(desc.getTaskId) + .setState(TaskState.TASK_FAILED) + .setData(ByteString.copyFrom(Utils.serialize(reason))) + .build()) } case t: Throwable => { val reason = ExceptionFailure(t) d.sendStatusUpdate(TaskStatus.newBuilder() - .setTaskId(desc.getTaskId) - .setState(TaskState.TASK_FAILED) - .setData(ByteString.copyFrom(Utils.serialize(reason))) - .build()) + .setTaskId(desc.getTaskId) + .setState(TaskState.TASK_FAILED) + .setData(ByteString.copyFrom(Utils.serialize(reason))) + .build()) // TODO: Handle errors in tasks less dramatically logError("Exception in task ID " + tid, t) @@ -102,8 +103,10 @@ class Executor extends org.apache.mesos.Executor with Logging { } } - // Create a ClassLoader for use in tasks, adding any JARs specified by the - // user or any classes created by the interpreter to the search path + /** + * Create a ClassLoader for use in tasks, adding any JARs specified by the user or any classes + * created by the interpreter to the search path + */ private def createClassLoader(): ClassLoader = { var loader = this.getClass.getClassLoader diff --git a/core/src/main/scala/spark/HadoopRDD.scala b/core/src/main/scala/spark/HadoopRDD.scala index 47286e0a65..62468d04d8 100644 --- a/core/src/main/scala/spark/HadoopRDD.scala +++ b/core/src/main/scala/spark/HadoopRDD.scala @@ -12,9 +12,15 @@ import org.apache.hadoop.mapred.RecordReader import org.apache.hadoop.mapred.Reporter import org.apache.hadoop.util.ReflectionUtils -/** A Spark split class that wraps around a Hadoop InputSplit */ -class HadoopSplit(rddId: Int, idx: Int, @transient s: InputSplit) -extends Split with Serializable { +/** + * A Spark split class that wraps around a Hadoop InputSplit. + */ +class HadoopSplit( + rddId: Int, + idx: Int, + @transient s: InputSplit + ) extends Split with Serializable { + val inputSplit = new SerializableWritable[InputSplit](s) override def hashCode(): Int = (41 * (41 + rddId) + idx).toInt @@ -22,10 +28,9 @@ extends Split with Serializable { override val index = idx } - /** - * An RDD that reads a Hadoop dataset as specified by a JobConf (e.g. files in - * HDFS, the local file system, or S3, tables in HBase, etc). + * An RDD that reads a Hadoop dataset as specified by a JobConf (e.g. files in HDFS, the local file + * system, or S3, tables in HBase, etc). */ class HadoopRDD[K, V]( sc: SparkContext, @@ -33,31 +38,37 @@ class HadoopRDD[K, V]( inputFormatClass: Class[_ <: InputFormat[K, V]], keyClass: Class[K], valueClass: Class[V], - minSplits: Int) -extends RDD[(K, V)](sc) { + minSplits: Int + ) extends RDD[(K, V)](sc) { + val serializableConf = new SerializableWritable(conf) - @transient val splits_ : Array[Split] = { + @transient + val splits_ : Array[Split] = { val inputFormat = createInputFormat(conf) val inputSplits = inputFormat.getSplits(conf, minSplits) - val array = new Array[Split] (inputSplits.size) - for (i <- 0 until inputSplits.size) + val array = new Array[Split](inputSplits.size) + for (i <- 0 until inputSplits.size) { array(i) = new HadoopSplit(id, i, inputSplits(i)) + } array } def createInputFormat(conf: JobConf): InputFormat[K, V] = { ReflectionUtils.newInstance(inputFormatClass.asInstanceOf[Class[_]], conf) - .asInstanceOf[InputFormat[K, V]] + .asInstanceOf[InputFormat[K, V]] } - // Helper method for creating a Hadoop Writable, because the commonly used - // NullWritable class has no constructor + /** + * Helper method for creating a Hadoop Writable, because the commonly used NullWritable class has + * no constructor. + */ def createWritable[T](clazz: Class[T]): T = { - if (clazz == classOf[NullWritable]) + if (clazz == classOf[NullWritable]) { NullWritable.get().asInstanceOf[T] - else + } else { clazz.newInstance() + } } override def splits = splits_ @@ -80,8 +91,7 @@ extends RDD[(K, V)](sc) { try { finished = !reader.next(key, value) } catch { - case eofe: java.io.EOFException => - finished = true + case eofe: java.io.EOFException => finished = true } gotNext = true } diff --git a/core/src/main/scala/spark/HadoopWriter.scala b/core/src/main/scala/spark/HadoopWriter.scala index 73c8876eb6..5574ffc28f 100644 --- a/core/src/main/scala/spark/HadoopWriter.scala +++ b/core/src/main/scala/spark/HadoopWriter.scala @@ -16,11 +16,14 @@ import spark.SerializableWritable import spark.Logging /** - * Saves an RDD using a Hadoop OutputFormat as specified by a JobConf. The JobConf should - * also contain an output key class, an output value class, a filename to write to, etc - * exactly like in a Hadoop job. + * Saves an RDD using a Hadoop OutputFormat as specified by a JobConf. The + * JobConf should also contain an output key class, an output value class, a + * filename to write to, etc exactly like in a Hadoop job. */ -class HadoopWriter(@transient jobConf: JobConf) extends Logging with Serializable { +class HadoopWriter( + @transient jobConf: JobConf + ) extends Logging with Serializable { + private val now = new Date() private val conf = new SerializableWritable(jobConf) @@ -58,22 +61,25 @@ class HadoopWriter(@transient jobConf: JobConf) extends Logging with Serializabl val outputName = "part-" + numfmt.format(splitID) val path = FileOutputFormat.getOutputPath(conf.value) val fs: FileSystem = { - if (path != null) + if (path != null) { path.getFileSystem(conf.value) - else + } else { FileSystem.get(conf.value) + } } getOutputCommitter().setupTask(getTaskContext()) - writer = getOutputFormat().getRecordWriter(fs, conf.value, outputName, Reporter.NULL) + writer = getOutputFormat().getRecordWriter( + fs, conf.value, outputName, Reporter.NULL) } def write(key: AnyRef, value: AnyRef) { if (writer!=null) { //println (">>> Writing ("+key.toString+": " + key.getClass.toString + ", " + value.toString + ": " + value.getClass.toString + ")") writer.write(key, value) - } else + } else { throw new IOException("Writer is null, open() has not been called") + } } def close() { @@ -109,26 +115,31 @@ class HadoopWriter(@transient jobConf: JobConf) extends Logging with Serializabl // ********* Private Functions ********* private def getOutputFormat(): OutputFormat[AnyRef,AnyRef] = { - if (format == null) - format = conf.value.getOutputFormat().asInstanceOf[OutputFormat[AnyRef,AnyRef]] + if (format == null) { + format = conf.value.getOutputFormat() + .asInstanceOf[OutputFormat[AnyRef,AnyRef]] + } return format } private def getOutputCommitter(): OutputCommitter = { - if (committer == null) + if (committer == null) { committer = conf.value.getOutputCommitter().asInstanceOf[OutputCommitter] + } return committer } private def getJobContext(): JobContext = { - if (jobContext == null) - jobContext = new JobContext(conf.value, jID.value) + if (jobContext == null) { + jobContext = new JobContext(conf.value, jID.value) + } return jobContext } private def getTaskContext(): TaskAttemptContext = { - if (taskContext == null) + if (taskContext == null) { taskContext = new TaskAttemptContext(conf.value, taID.value) + } return taskContext } @@ -158,12 +169,14 @@ object HadoopWriter { } def createPathFromString(path: String, conf: JobConf): Path = { - if (path == null) + if (path == null) { throw new IllegalArgumentException("Output path is null") + } var outputPath = new Path(path) val fs = outputPath.getFileSystem(conf) - if (outputPath == null || fs == null) + if (outputPath == null || fs == null) { throw new IllegalArgumentException("Incorrectly formatted output path") + } outputPath = outputPath.makeQualified(fs) return outputPath } diff --git a/core/src/main/scala/spark/HttpServer.scala b/core/src/main/scala/spark/HttpServer.scala index d2a663ac1f..855f2c752f 100644 --- a/core/src/main/scala/spark/HttpServer.scala +++ b/core/src/main/scala/spark/HttpServer.scala @@ -9,18 +9,15 @@ import org.eclipse.jetty.server.handler.HandlerList import org.eclipse.jetty.server.handler.ResourceHandler import org.eclipse.jetty.util.thread.QueuedThreadPool - /** - * Exception type thrown by HttpServer when it is in the wrong state - * for an operation. + * Exception type thrown by HttpServer when it is in the wrong state for an operation. */ class ServerStateException(message: String) extends Exception(message) - /** - * An HTTP server for static content used to allow worker nodes to access JARs - * added to SparkContext as well as classes created by the interpreter when - * the user types in code. This is just a wrapper around a Jetty server. + * An HTTP server for static content used to allow worker nodes to access JARs added to SparkContext + * as well as classes created by the interpreter when the user types in code. This is just a wrapper + * around a Jetty server. */ class HttpServer(resourceBase: File) extends Logging { private var server: Server = null diff --git a/core/src/main/scala/spark/Job.scala b/core/src/main/scala/spark/Job.scala index 2200fb0c5d..9846e91873 100644 --- a/core/src/main/scala/spark/Job.scala +++ b/core/src/main/scala/spark/Job.scala @@ -4,8 +4,8 @@ import org.apache.mesos._ import org.apache.mesos.Protos._ /** - * Class representing a parallel job in MesosScheduler. Schedules the - * job by implementing various callbacks. + * Class representing a parallel job in MesosScheduler. Schedules the job by implementing various + * callbacks. */ abstract class Job(jobId: Int) { def slaveOffer(s: Offer, availableCpus: Double): Option[TaskDescription] diff --git a/core/src/main/scala/spark/LocalScheduler.scala b/core/src/main/scala/spark/LocalScheduler.scala index 6485da0b51..6a66d9deb8 100644 --- a/core/src/main/scala/spark/LocalScheduler.scala +++ b/core/src/main/scala/spark/LocalScheduler.scala @@ -4,9 +4,9 @@ import java.util.concurrent.Executors import java.util.concurrent.atomic.AtomicInteger /** - * A simple Scheduler implementation that runs tasks locally in a thread pool. - * Optionally the scheduler also allows each task to fail up to maxFailures times, - * which is useful for testing fault recovery. + * A simple Scheduler implementation that runs tasks locally in a thread pool. Optionally the + * scheduler also allows each task to fail up to maxFailures times, which is useful for testing + * fault recovery. */ private class LocalScheduler(threads: Int, maxFailures: Int) extends DAGScheduler with Logging { var attemptId = new AtomicInteger(0) @@ -35,9 +35,8 @@ private class LocalScheduler(threads: Int, maxFailures: Int) extends DAGSchedule // Set the Spark execution environment for the worker thread SparkEnv.set(env) try { - // Serialize and deserialize the task so that accumulators are - // changed to thread-local ones; this adds a bit of unnecessary - // overhead but matches how the Mesos Executor works + // Serialize and deserialize the task so that accumulators are changed to thread-local ones; + // this adds a bit of unnecessary overhead but matches how the Mesos Executor works Accumulators.clear val bytes = Utils.serialize(task) logInfo("Size of task " + idInJob + " is " + bytes.size + " bytes") diff --git a/core/src/main/scala/spark/MesosScheduler.scala b/core/src/main/scala/spark/MesosScheduler.scala index c9c2f169f6..3854915852 100644 --- a/core/src/main/scala/spark/MesosScheduler.scala +++ b/core/src/main/scala/spark/MesosScheduler.scala @@ -19,13 +19,15 @@ import org.apache.mesos._ import org.apache.mesos.Protos._ /** - * The main Scheduler implementation, which runs jobs on Mesos. Clients should - * first call start(), then submit tasks through the runTasks method. + * The main Scheduler implementation, which runs jobs on Mesos. Clients should first call start(), + * then submit tasks through the runTasks method. */ private class MesosScheduler( - sc: SparkContext, master: String, frameworkName: String) -extends MScheduler with DAGScheduler with Logging -{ + sc: SparkContext, + master: String, + frameworkName: String + )extends MScheduler with DAGScheduler with Logging { + // Environment variables to pass to our executors val ENV_VARS_TO_SEND_TO_EXECUTORS = Array( "SPARK_MEM", @@ -36,14 +38,15 @@ extends MScheduler with DAGScheduler with Logging // Memory used by each executor (in megabytes) val EXECUTOR_MEMORY = { - if (System.getenv("SPARK_MEM") != null) + if (System.getenv("SPARK_MEM") != null) { memoryStringToMb(System.getenv("SPARK_MEM")) // TODO: Might need to add some extra memory for the non-heap parts of the JVM - else + } else { 512 + } } - // Lock used to wait for scheduler to be registered + // Lock used to wait for scheduler to be registered private var isRegistered = false private val registeredLock = new Object() @@ -92,13 +95,13 @@ extends MScheduler with DAGScheduler with Logging setDaemon(true) override def run { val sched = MesosScheduler.this - driver = new MesosSchedulerDriver(sched, frameworkName, getExecutorInfo, master) + driver = new MesosSchedulerDriver( + sched, frameworkName, getExecutorInfo, master) try { val ret = driver.run() logInfo("driver.run() returned with code " + ret) } catch { - case e: Exception => - logError("driver.run() failed", e) + case e: Exception => logError("driver.run() failed", e) } } }.start @@ -117,17 +120,16 @@ extends MScheduler with DAGScheduler with Logging for (key <- ENV_VARS_TO_SEND_TO_EXECUTORS) { if (System.getenv(key) != null) { params.addParam(Param.newBuilder() - .setKey("env." + key) - .setValue(System.getenv(key)) - .build()) + .setKey("env." + key) + .setValue(System.getenv(key)) + .build()) } } val memory = Resource.newBuilder() - .setName("mem") - .setType(Resource.Type.SCALAR) - .setScalar(Resource.Scalar.newBuilder() - .setValue(EXECUTOR_MEMORY).build()) - .build() + .setName("mem") + .setType(Resource.Type.SCALAR) + .setScalar(Resource.Scalar.newBuilder().setValue(EXECUTOR_MEMORY).build()) + .build() ExecutorInfo.newBuilder() .setExecutorId(ExecutorID.newBuilder().setValue("default").build()) .setUri(execScript) @@ -178,9 +180,9 @@ extends MScheduler with DAGScheduler with Logging } /** - * Method called by Mesos to offer resources on slaves. We resond by asking - * our active jobs for tasks in FIFO order. We fill each node with tasks in - * a round-robin manner so that tasks are balanced across the cluster. + * Method called by Mesos to offer resources on slaves. We resond by asking our active jobs for + * tasks in FIFO order. We fill each node with tasks in a round-robin manner so that tasks are + * balanced across the cluster. */ override def resourceOffers(d: SchedulerDriver, offers: JList[Offer]) { synchronized { @@ -238,7 +240,8 @@ extends MScheduler with DAGScheduler with Logging synchronized { try { val tid = status.getTaskId.getValue - if (status.getState == TaskState.TASK_LOST && taskIdToSlaveId.contains(tid)) { + if (status.getState == TaskState.TASK_LOST + && taskIdToSlaveId.contains(tid)) { // We lost the executor on this slave, so remember that it's gone slavesWithExecutors -= taskIdToSlaveId(tid) } @@ -249,8 +252,9 @@ extends MScheduler with DAGScheduler with Logging } if (isFinished(status.getState)) { taskIdToJobId.remove(tid) - if (jobTasks.contains(jobId)) + if (jobTasks.contains(jobId)) { jobTasks(jobId) -= tid + } taskIdToSlaveId.remove(tid) } case None => @@ -346,7 +350,10 @@ extends MScheduler with DAGScheduler with Logging return Utils.serialize(props.toArray) } - override def frameworkMessage(d: SchedulerDriver, s: SlaveID, e: ExecutorID, b: Array[Byte]) {} + override def frameworkMessage(d: SchedulerDriver, + s: SlaveID, + e: ExecutorID, + b: Array[Byte]) {} override def slaveLost(d: SchedulerDriver, s: SlaveID) { slavesWithExecutors.remove(s.getValue) @@ -361,15 +368,16 @@ extends MScheduler with DAGScheduler with Logging */ def memoryStringToMb(str: String): Int = { val lower = str.toLowerCase - if (lower.endsWith("k")) + if (lower.endsWith("k")) { (lower.substring(0, lower.length-1).toLong / 1024).toInt - else if (lower.endsWith("m")) + } else if (lower.endsWith("m")) { lower.substring(0, lower.length-1).toInt - else if (lower.endsWith("g")) + } else if (lower.endsWith("g")) { lower.substring(0, lower.length-1).toInt * 1024 - else if (lower.endsWith("t")) + } else if (lower.endsWith("t")) { lower.substring(0, lower.length-1).toInt * 1024 * 1024 - else // no suffix, so it's just a number in bytes - (lower.toLong / 1024 / 1024).toInt + } else {// no suffix, so it's just a number in bytes + (lower.toLong / 1024 / 1024).toInt + } } } diff --git a/core/src/main/scala/spark/ParallelCollection.scala b/core/src/main/scala/spark/ParallelCollection.scala index e96f73b3cf..4bcb9e0d54 100644 --- a/core/src/main/scala/spark/ParallelCollection.scala +++ b/core/src/main/scala/spark/ParallelCollection.scala @@ -4,15 +4,17 @@ import scala.collection.immutable.NumericRange import scala.collection.mutable.ArrayBuffer class ParallelCollectionSplit[T: ClassManifest]( - val rddId: Long, val slice: Int, values: Seq[T]) -extends Split with Serializable { + val rddId: Long, + val slice: Int, + values: Seq[T] + ) extends Split with Serializable { + def iterator(): Iterator[T] = values.iterator override def hashCode(): Int = (41 * (41 + rddId) + slice).toInt override def equals(other: Any): Boolean = other match { - case that: ParallelCollectionSplit[_] => - (this.rddId == that.rddId && this.slice == that.slice) + case that: ParallelCollectionSplit[_] => (this.rddId == that.rddId && this.slice == that.slice) case _ => false } @@ -20,13 +22,16 @@ extends Split with Serializable { } class ParallelCollection[T: ClassManifest]( - sc: SparkContext, @transient data: Seq[T], numSlices: Int) -extends RDD[T](sc) { - // TODO: Right now, each split sends along its full data, even if later down - // the RDD chain it gets cached. It might be worthwhile to write the data to - // a file in the DFS and read it in the split instead. + sc: SparkContext, + @transient data: Seq[T], + numSlices: Int + ) extends RDD[T](sc) { + // TODO: Right now, each split sends along its full data, even if later down the RDD chain it gets + // cached. It might be worthwhile to write the data to a file in the DFS and read it in the split + // instead. - @transient val splits_ = { + @transient + val splits_ = { val slices = ParallelCollection.slice(data, numSlices).toArray slices.indices.map(i => new ParallelCollectionSplit(id, i, slices(i))).toArray } @@ -41,17 +46,24 @@ extends RDD[T](sc) { } private object ParallelCollection { - // Slice a collection into numSlices sub-collections. One extra thing we do here is - // to treat Range collections specially, encoding the slices as other Ranges to - // minimize memory cost. This makes it efficient to run Spark over RDDs representing - // large sets of numbers. + /** + * Slice a collection into numSlices sub-collections. One extra thing we do here is to treat Range + * collections specially, encoding the slices as other Ranges to minimize memory cost. This makes + * it efficient to run Spark over RDDs representing large sets of numbers. + */ def slice[T: ClassManifest](seq: Seq[T], numSlices: Int): Seq[Seq[T]] = { - if (numSlices < 1) + if (numSlices < 1) { throw new IllegalArgumentException("Positive number of slices required") + } seq match { case r: Range.Inclusive => { - val sign = if (r.step < 0) -1 else 1 - slice(new Range(r.start, r.end + sign, r.step).asInstanceOf[Seq[T]], numSlices) + val sign = if (r.step < 0) { + -1 + } else { + 1 + } + slice(new Range( + r.start, r.end + sign, r.step).asInstanceOf[Seq[T]], numSlices) } case r: Range => { (0 until numSlices).map(i => { diff --git a/core/src/main/scala/spark/RDD.scala b/core/src/main/scala/spark/RDD.scala index e186ddae82..5ba3d1b7ea 100644 --- a/core/src/main/scala/spark/RDD.scala +++ b/core/src/main/scala/spark/RDD.scala @@ -27,24 +27,26 @@ import org.apache.hadoop.mapred.TextOutputFormat import SparkContext._ /** - * A Resilient Distributed Dataset (RDD), the basic abstraction in Spark. Represents - * an immutable, partitioned collection of elements that can be operated on in parallel. + * A Resilient Distributed Dataset (RDD), the basic abstraction in Spark. Represents an immutable, + * partitioned collection of elements that can be operated on in parallel. * * Each RDD is characterized by five main properties: * - A list of splits (partitions) * - A function for computing each split * - A list of dependencies on other RDDs * - Optionally, a Partitioner for key-value RDDs (e.g. to say that the RDD is hash-partitioned) - * - Optionally, a list of preferred locations to compute each split on (e.g. block locations for HDFS) + * - Optionally, a list of preferred locations to compute each split on (e.g. block locations for + * HDFS) * - * All the scheduling and execution in Spark is done based on these methods, allowing each - * RDD to implement its own way of computing itself. + * All the scheduling and execution in Spark is done based on these methods, allowing each RDD to + * implement its own way of computing itself. * - * This class also contains transformation methods available on all RDDs (e.g. map and filter). - * In addition, PairRDDFunctions contains extra methods available on RDDs of key-value pairs, - * and SequenceFileRDDFunctions contains extra methods for saving RDDs to Hadoop SequenceFiles. + * This class also contains transformation methods available on all RDDs (e.g. map and filter). In + * addition, PairRDDFunctions contains extra methods available on RDDs of key-value pairs, and + * SequenceFileRDDFunctions contains extra methods for saving RDDs to Hadoop SequenceFiles. */ abstract class RDD[T: ClassManifest](@transient sc: SparkContext) extends Serializable { + // Methods that must be implemented by subclasses def splits: Array[Split] def compute(split: Split): Iterator[T] @@ -100,19 +102,16 @@ abstract class RDD[T: ClassManifest](@transient sc: SparkContext) extends Serial if (initialCount > Integer.MAX_VALUE) { maxSelected = Integer.MAX_VALUE - } - else { + } else { maxSelected = initialCount.toInt } if (num > initialCount) { total = maxSelected fraction = Math.min(multiplier*(maxSelected+1)/initialCount, 1.0) - } - else if (num < 0) { + } else if (num < 0) { throw(new IllegalArgumentException("Negative number of elements requested")) - } - else { + } else { fraction = Math.min(multiplier*(num+1)/initialCount, 1.0) total = num.toInt } @@ -134,22 +133,18 @@ abstract class RDD[T: ClassManifest](@transient sc: SparkContext) extends Serial def glom(): RDD[Array[T]] = new GlommedRDD(this) - def cartesian[U: ClassManifest](other: RDD[U]): RDD[(T, U)] = - new CartesianRDD(sc, this, other) + def cartesian[U: ClassManifest](other: RDD[U]): RDD[(T, U)] = new CartesianRDD(sc, this, other) def groupBy[K: ClassManifest](f: T => K, numSplits: Int): RDD[(K, Seq[T])] = { val cleanF = sc.clean(f) this.map(t => (cleanF(t), t)).groupByKey(numSplits) } - def groupBy[K: ClassManifest](f: T => K): RDD[(K, Seq[T])] = - groupBy[K](f, sc.defaultParallelism) + def groupBy[K: ClassManifest](f: T => K): RDD[(K, Seq[T])] = groupBy[K](f, sc.defaultParallelism) - def pipe(command: String): RDD[String] = - new PipedRDD(this, command) + def pipe(command: String): RDD[String] = new PipedRDD(this, command) - def pipe(command: Seq[String]): RDD[String] = - new PipedRDD(this, command) + def pipe(command: Seq[String]): RDD[String] = new PipedRDD(this, command) def mapPartitions[U: ClassManifest](f: Iterator[T] => Iterator[U]): RDD[U] = new MapPartitionsRDD(this, sc.clean(f)) @@ -169,26 +164,29 @@ abstract class RDD[T: ClassManifest](@transient sc: SparkContext) extends Serial def reduce(f: (T, T) => T): T = { val cleanF = sc.clean(f) val reducePartition: Iterator[T] => Option[T] = iter => { - if (iter.hasNext) + if (iter.hasNext) { Some(iter.reduceLeft(cleanF)) - else + }else { None + } } val options = sc.runJob(this, reducePartition) val results = new ArrayBuffer[T] - for (opt <- options; elem <- opt) + for (opt <- options; elem <- opt) { results += elem - if (results.size == 0) + } + if (results.size == 0) { throw new UnsupportedOperationException("empty collection") - else + } else { return results.reduceLeft(cleanF) + } } /** - * Aggregate the elements of each partition, and then the results for all the - * partitions, using a given associative function and a neutral "zero value". - * The function op(t1, t2) is allowed to modify t1 and return it as its result - * value to avoid object allocation; however, it should not modify t2. + * Aggregate the elements of each partition, and then the results for all the partitions, using a + * given associative function and a neutral "zero value". The function op(t1, t2) is allowed to + * modify t1 and return it as its result value to avoid object allocation; however, it should not + * modify t2. */ def fold(zeroValue: T)(op: (T, T) => T): T = { val cleanOp = sc.clean(op) @@ -197,19 +195,20 @@ abstract class RDD[T: ClassManifest](@transient sc: SparkContext) extends Serial } /** - * Aggregate the elements of each partition, and then the results for all the - * partitions, using given combine functions and a neutral "zero value". This - * function can return a different result type, U, than the type of this RDD, T. - * Thus, we need one operation for merging a T into an U and one operation for - * merging two U's, as in scala.TraversableOnce. Both of these functions are - * allowed to modify and return their first argument instead of creating a new U - * to avoid memory allocation. + * Aggregate the elements of each partition, and then the results for all the partitions, using + * given combine functions and a neutral "zero value". This function can return a different result + * type, U, than the type of this RDD, T. Thus, we need one operation for merging a T into an U + * and one operation for merging two U's, as in scala.TraversableOnce. Both of these functions are + * allowed to modify and return their first argument instead of creating a new U to avoid memory + * allocation. */ - def aggregate[U: ClassManifest](zeroValue: U)(seqOp: (U, T) => U, combOp: (U, U) => U): U = { + def aggregate[U: ClassManifest](zeroValue: U)( + seqOp: (U, T) => U, + combOp: (U, U) => U): U = { val cleanSeqOp = sc.clean(seqOp) val cleanCombOp = sc.clean(combOp) val results = sc.runJob(this, - (iter: Iterator[T]) => iter.aggregate(zeroValue)(cleanSeqOp, cleanCombOp)) + (iter: Iterator[T]) => iter.aggregate(zeroValue)(cleanSeqOp, cleanCombOp)) return results.fold(zeroValue)(cleanCombOp) } @@ -226,12 +225,15 @@ abstract class RDD[T: ClassManifest](@transient sc: SparkContext) extends Serial def toArray(): Array[T] = collect() - // Take the first num elements of the RDD. This currently scans the partitions - // *one by one*, so it will be slow if a lot of partitions are required. In that - // case, use collect() to get the whole RDD instead. + /** + * Take the first num elements of the RDD. This currently scans the partitions *one by one*, so + * it will be slow if a lot of partitions are required. In that case, use collect() to get the + * whole RDD instead. + */ def take(num: Int): Array[T] = { - if (num == 0) + if (num == 0) { return new Array[T](0) + } val buf = new ArrayBuffer[T] var p = 0 while (buf.size < num && p < splits.size) { @@ -251,48 +253,57 @@ abstract class RDD[T: ClassManifest](@transient sc: SparkContext) extends Serial } def saveAsTextFile(path: String) { - this.map(x => (NullWritable.get(), new Text(x.toString))).saveAsHadoopFile[TextOutputFormat[NullWritable, Text]](path) + this.map(x => (NullWritable.get(), new Text(x.toString))) + .saveAsHadoopFile[TextOutputFormat[NullWritable, Text]](path) } def saveAsObjectFile(path: String) { - this.glom.map(x => (NullWritable.get(), new BytesWritable(Utils.serialize(x)))).saveAsSequenceFile(path) + this.glom + .map(x => (NullWritable.get(), new BytesWritable(Utils.serialize(x)))) + .saveAsSequenceFile(path) } } class MappedRDD[U: ClassManifest, T: ClassManifest]( - prev: RDD[T], f: T => U) -extends RDD[U](prev.context) { + prev: RDD[T], + f: T => U + ) extends RDD[U](prev.context) { override def splits = prev.splits override val dependencies = List(new OneToOneDependency(prev)) override def compute(split: Split) = prev.iterator(split).map(f) } class FlatMappedRDD[U: ClassManifest, T: ClassManifest]( - prev: RDD[T], f: T => Traversable[U]) -extends RDD[U](prev.context) { + prev: RDD[T], + f: T => Traversable[U] + ) extends RDD[U](prev.context) { override def splits = prev.splits override val dependencies = List(new OneToOneDependency(prev)) override def compute(split: Split) = prev.iterator(split).flatMap(f) } class FilteredRDD[T: ClassManifest]( - prev: RDD[T], f: T => Boolean) -extends RDD[T](prev.context) { + prev: RDD[T], + f: T => Boolean + ) extends RDD[T](prev.context) { override def splits = prev.splits override val dependencies = List(new OneToOneDependency(prev)) - override def compute(split: Split) = prev.iterator(split).filter(f) + override def compute(split: Split) = + prev.iterator(split).filter(f) } -class GlommedRDD[T: ClassManifest](prev: RDD[T]) -extends RDD[Array[T]](prev.context) { +class GlommedRDD[T: ClassManifest]( + prev: RDD[T] + ) extends RDD[Array[T]](prev.context) { override def splits = prev.splits override val dependencies = List(new OneToOneDependency(prev)) override def compute(split: Split) = Array(prev.iterator(split).toArray).iterator } class MapPartitionsRDD[U: ClassManifest, T: ClassManifest]( - prev: RDD[T], f: Iterator[T] => Iterator[U]) -extends RDD[U](prev.context) { + prev: RDD[T], + f: Iterator[T] => Iterator[U] + ) extends RDD[U](prev.context) { override def splits = prev.splits override val dependencies = List(new OneToOneDependency(prev)) override def compute(split: Split) = f(prev.iterator(split)) diff --git a/core/src/main/scala/spark/SampledRDD.scala b/core/src/main/scala/spark/SampledRDD.scala index 21c1148b63..89d91e5603 100644 --- a/core/src/main/scala/spark/SampledRDD.scala +++ b/core/src/main/scala/spark/SampledRDD.scala @@ -7,16 +7,24 @@ class SampledRDDSplit(val prev: Split, val seed: Int) extends Split with Seriali } class SampledRDD[T: ClassManifest]( - prev: RDD[T], withReplacement: Boolean, frac: Double, seed: Int) -extends RDD[T](prev.context) { + prev: RDD[T], + withReplacement: Boolean, + frac: Double, + seed: Int + ) extends RDD[T](prev.context) { - @transient val splits_ = { val rg = new Random(seed); prev.splits.map(x => new SampledRDDSplit(x, rg.nextInt)) } + @transient + val splits_ = { + val rg = new Random(seed); + prev.splits.map(x => new SampledRDDSplit(x, rg.nextInt)) + } override def splits = splits_.asInstanceOf[Array[Split]] override val dependencies = List(new OneToOneDependency(prev)) - override def preferredLocations(split: Split) = prev.preferredLocations(split.asInstanceOf[SampledRDDSplit].prev) + override def preferredLocations(split: Split) = + prev.preferredLocations(split.asInstanceOf[SampledRDDSplit].prev) override def compute(splitIn: Split) = { val split = splitIn.asInstanceOf[SampledRDDSplit] @@ -25,11 +33,13 @@ extends RDD[T](prev.context) { if (withReplacement) { val oldData = prev.iterator(split.prev).toArray val sampleSize = (oldData.size * frac).ceil.toInt - val sampledData = for (i <- 1 to sampleSize) yield oldData(rg.nextInt(oldData.size)) // all of oldData's indices are candidates, even if sampleSize < oldData.size + val sampledData = { + // all of oldData's indices are candidates, even if sampleSize < oldData.size + for (i <- 1 to sampleSize) + yield oldData(rg.nextInt(oldData.size)) + } sampledData.iterator - } - // Sampling without replacement - else { + } else { // Sampling without replacement prev.iterator(split.prev).filter(x => (rg.nextDouble <= frac)) } } diff --git a/core/src/main/scala/spark/Scheduler.scala b/core/src/main/scala/spark/Scheduler.scala index df86db64a6..6c7e569313 100644 --- a/core/src/main/scala/spark/Scheduler.scala +++ b/core/src/main/scala/spark/Scheduler.scala @@ -1,17 +1,24 @@ package spark -// Scheduler trait, implemented by both MesosScheduler and LocalScheduler. +/** + * Scheduler trait, implemented by both MesosScheduler and LocalScheduler. + */ private trait Scheduler { def start() // Wait for registration with Mesos. def waitForRegister() - // Run a function on some partitions of an RDD, returning an array of results. The allowLocal flag specifies - // whether the scheduler is allowed to run the job on the master machine rather than shipping it to the cluster, - // for actions that create short jobs such as first() and take(). - def runJob[T, U: ClassManifest](rdd: RDD[T], func: (TaskContext, Iterator[T]) => U, - partitions: Seq[Int], allowLocal: Boolean): Array[U] + /** + * Run a function on some partitions of an RDD, returning an array of results. The allowLocal + * flag specifies whether the scheduler is allowed to run the job on the master machine rather + * than shipping it to the cluster, for actions that create short jobs such as first() and take(). + */ + def runJob[T, U: ClassManifest]( + rdd: RDD[T], + func: (TaskContext, Iterator[T]) => U, + partitions: Seq[Int], + allowLocal: Boolean): Array[U] def stop() diff --git a/core/src/main/scala/spark/SimpleJob.scala b/core/src/main/scala/spark/SimpleJob.scala index bf881fb2d4..636e18eb4c 100644 --- a/core/src/main/scala/spark/SimpleJob.scala +++ b/core/src/main/scala/spark/SimpleJob.scala @@ -61,7 +61,8 @@ extends Job(jobId) with Logging var causeOfFailure = "" // How frequently to reprint duplicate exceptions in full, in milliseconds - val EXCEPTION_PRINT_INTERVAL = System.getProperty("spark.logging.exceptionPrintInterval", "10000").toLong + val EXCEPTION_PRINT_INTERVAL = + System.getProperty("spark.logging.exceptionPrintInterval", "10000").toLong // Map of recent exceptions (identified by string representation and // top stack frame) to duplicate count (how many times the same // exception has appeared) and time the full exception was @@ -171,12 +172,12 @@ extends Job(jobId) with Logging logDebug("Serialized size: " + serializedTask.size) val taskName = "task %d:%d".format(jobId, index) return Some(TaskDescription.newBuilder() - .setTaskId(taskId) - .setSlaveId(offer.getSlaveId) - .setName(taskName) - .addResources(cpuRes) - .setData(ByteString.copyFrom(serializedTask)) - .build()) + .setTaskId(taskId) + .setSlaveId(offer.getSlaveId) + .setName(taskName) + .addResources(cpuRes) + .setData(ByteString.copyFrom(serializedTask)) + .build()) } case _ => } diff --git a/core/src/main/scala/spark/SparkContext.scala b/core/src/main/scala/spark/SparkContext.scala index b0cc0e6454..4a0732bd5a 100644 --- a/core/src/main/scala/spark/SparkContext.scala +++ b/core/src/main/scala/spark/SparkContext.scala @@ -31,29 +31,32 @@ import org.apache.hadoop.mapreduce.{Job => NewHadoopJob} import spark.broadcast._ class SparkContext( - master: String, - frameworkName: String, - val sparkHome: String = null, - val jars: Seq[String] = Nil) -extends Logging { + master: String, + frameworkName: String, + val sparkHome: String = null, + val jars: Seq[String] = Nil + ) extends Logging { // Ensure logging is initialized before we spawn any threads initLogging() // Set Spark master host and port system properties - if (System.getProperty("spark.master.host") == null) + if (System.getProperty("spark.master.host") == null) { System.setProperty("spark.master.host", Utils.localHostName) - if (System.getProperty("spark.master.port") == null) + } + if (System.getProperty("spark.master.port") == null) { System.setProperty("spark.master.port", "7077") + } // Make sure a proper class loader is set for remote actors (unless user set one) - if (RemoteActor.classLoader == null) + if (RemoteActor.classLoader == null) { RemoteActor.classLoader = getClass.getClassLoader + } // Create the Spark execution environment (cache, map output tracker, etc) val env = SparkEnv.createFromSystemProperties(true) SparkEnv.set(env) Broadcast.initialize(true) - + // Create and start the scheduler private var scheduler: Scheduler = { // Regular expression used for local[N] master format @@ -61,10 +64,8 @@ extends Logging { // Regular expression for local[N, maxRetries], used in tests with failing tasks val LOCAL_N_FAILURES_REGEX = """local\[([0-9]+),([0-9]+)\]""".r master match { - case "local" => - new LocalScheduler(1, 0) - case LOCAL_N_REGEX(threads) => - new LocalScheduler(threads.toInt, 0) + case "local" => new LocalScheduler(1, 0) + case LOCAL_N_REGEX(threads) => new LocalScheduler(threads.toInt, 0) case LOCAL_N_FAILURES_REGEX(threads, maxFailures) => new LocalScheduler(threads.toInt, maxFailures.toInt) case _ => @@ -78,11 +79,19 @@ extends Logging { // Methods for creating RDDs - def parallelize[T: ClassManifest](seq: Seq[T], numSlices: Int = defaultParallelism): RDD[T] = + def parallelize[T: ClassManifest]( + seq: Seq[T], + numSlices: Int = defaultParallelism + ): RDD[T] = { new ParallelCollection[T](this, seq, numSlices) + } - def makeRDD[T: ClassManifest](seq: Seq[T], numSlices: Int = defaultParallelism): RDD[T] = + def makeRDD[T: ClassManifest]( + seq: Seq[T], + numSlices: Int = defaultParallelism + ): RDD[T] = { parallelize(seq, numSlices) + } def textFile(path: String, minSplits: Int = defaultMinSplits): RDD[String] = { hadoopFile(path, classOf[TextInputFormat], classOf[LongWritable], classOf[Text], minSplits) @@ -90,26 +99,28 @@ extends Logging { } /** - * Get an RDD for a Hadoop-readable dataset from a Hadooop JobConf giving - * its InputFormat and any other necessary info (e.g. file name for a - * filesystem-based dataset, table name for HyperTable, etc). + * Get an RDD for a Hadoop-readable dataset from a Hadooop JobConf giving its InputFormat and any + * other necessary info (e.g. file name for a filesystem-based dataset, table name for HyperTable, + * etc). */ - def hadoopRDD[K, V](conf: JobConf, - inputFormatClass: Class[_ <: InputFormat[K, V]], - keyClass: Class[K], - valueClass: Class[V], - minSplits: Int = defaultMinSplits) - : RDD[(K, V)] = { + def hadoopRDD[K, V]( + conf: JobConf, + inputFormatClass: Class[_ <: InputFormat[K, V]], + keyClass: Class[K], + valueClass: Class[V], + minSplits: Int = defaultMinSplits + ): RDD[(K, V)] = { new HadoopRDD(this, conf, inputFormatClass, keyClass, valueClass, minSplits) } /** Get an RDD for a Hadoop file with an arbitrary InputFormat */ - def hadoopFile[K, V](path: String, - inputFormatClass: Class[_ <: InputFormat[K, V]], - keyClass: Class[K], - valueClass: Class[V], - minSplits: Int = defaultMinSplits) - : RDD[(K, V)] = { + def hadoopFile[K, V]( + path: String, + inputFormatClass: Class[_ <: InputFormat[K, V]], + keyClass: Class[K], + valueClass: Class[V], + minSplits: Int = defaultMinSplits + ) : RDD[(K, V)] = { val conf = new JobConf() FileInputFormat.setInputPaths(conf, path) val bufferSize = System.getProperty("spark.buffer.size", "65536") @@ -118,15 +129,17 @@ extends Logging { } /** - * Smarter version of hadoopFile() that uses class manifests to figure out - * the classes of keys, values and the InputFormat so that users don't need - * to pass them directly. + * Smarter version of hadoopFile() that uses class manifests to figure out the classes of keys, + * values and the InputFormat so that users don't need to pass them directly. */ def hadoopFile[K, V, F <: InputFormat[K, V]](path: String, minSplits: Int) (implicit km: ClassManifest[K], vm: ClassManifest[V], fm: ClassManifest[F]) : RDD[(K, V)] = { - hadoopFile(path, fm.erasure.asInstanceOf[Class[F]], km.erasure.asInstanceOf[Class[K]], - vm.erasure.asInstanceOf[Class[V]], minSplits) + hadoopFile(path, + fm.erasure.asInstanceOf[Class[F]], + km.erasure.asInstanceOf[Class[K]], + vm.erasure.asInstanceOf[Class[V]], + minSplits) } def hadoopFile[K, V, F <: InputFormat[K, V]](path: String) @@ -136,31 +149,34 @@ extends Logging { /** Get an RDD for a Hadoop file with an arbitrary new API InputFormat. */ def newAPIHadoopFile[K, V, F <: NewInputFormat[K, V]](path: String) (implicit km: ClassManifest[K], vm: ClassManifest[V], fm: ClassManifest[F]): RDD[(K, V)] = { - val job = new NewHadoopJob - NewFileInputFormat.addInputPath(job, new Path(path)) - val conf = job.getConfiguration - newAPIHadoopFile(path, - fm.erasure.asInstanceOf[Class[F]], - km.erasure.asInstanceOf[Class[K]], - vm.erasure.asInstanceOf[Class[V]], - conf) + val job = new NewHadoopJob + NewFileInputFormat.addInputPath(job, new Path(path)) + val conf = job.getConfiguration + newAPIHadoopFile(path, + fm.erasure.asInstanceOf[Class[F]], + km.erasure.asInstanceOf[Class[K]], + vm.erasure.asInstanceOf[Class[V]], + conf) } - /** Get an RDD for a given Hadoop file with an arbitrary new API InputFormat and extra - * configuration options to pass to the input format. + /** + * Get an RDD for a given Hadoop file with an arbitrary new API InputFormat + * and extra configuration options to pass to the input format. */ - def newAPIHadoopFile[K, V, F <: NewInputFormat[K, V]](path: String, - fClass: Class[F], - kClass: Class[K], - vClass: Class[V], - conf: Configuration): RDD[(K, V)] = - new NewHadoopRDD(this, fClass, kClass, vClass, conf) + def newAPIHadoopFile[K, V, F <: NewInputFormat[K, V]]( + path: String, + fClass: Class[F], + kClass: Class[K], + vClass: Class[V], + conf: Configuration + ): RDD[(K, V)] = new NewHadoopRDD(this, fClass, kClass, vClass, conf) /** Get an RDD for a Hadoop SequenceFile with given key and value types */ def sequenceFile[K, V](path: String, - keyClass: Class[K], - valueClass: Class[V], - minSplits: Int): RDD[(K, V)] = { + keyClass: Class[K], + valueClass: Class[V], + minSplits: Int + ): RDD[(K, V)] = { val inputFormatClass = classOf[SequenceFileInputFormat[K, V]] hadoopFile(path, inputFormatClass, keyClass, valueClass, minSplits) } @@ -169,41 +185,49 @@ extends Logging { sequenceFile(path, keyClass, valueClass, defaultMinSplits) /** - * Version of sequenceFile() for types implicitly convertible to Writables through a WritableConverter. + * Version of sequenceFile() for types implicitly convertible to Writables + * through a WritableConverter. * - * WritableConverters are provided in a somewhat strange way (by an implicit function) to support both - * subclasses of Writable and types for which we define a converter (e.g. Int to IntWritable). The most - * natural thing would've been to have implicit objects for the converters, but then we couldn't have - * an object for every subclass of Writable (you can't have a parameterized singleton object). We use - * functions instead to create a new converter for the appropriate type. In addition, we pass the converter - * a ClassManifest of its type to allow it to figure out the Writable class to use in the subclass case. + * WritableConverters are provided in a somewhat strange way (by an implicit + * function) to support both subclasses of Writable and types for which we + * define a converter (e.g. Int to IntWritable). The most natural thing + * would've been to have implicit objects for the converters, but then we + * couldn't have an object for every subclass of Writable (you can't have a + * parameterized singleton object). We use functions instead to create a new + * converter for the appropriate type. In addition, we pass the converter a + * ClassManifest of its type to allow it to figure out the Writable class to + * use in the subclass case. */ def sequenceFile[K, V](path: String, minSplits: Int = defaultMinSplits) - (implicit km: ClassManifest[K], vm: ClassManifest[V], kcf: () => WritableConverter[K], vcf: () => WritableConverter[V]) + (implicit km: ClassManifest[K], vm: ClassManifest[V], + kcf: () => WritableConverter[K], vcf: () => WritableConverter[V]) : RDD[(K, V)] = { val kc = kcf() val vc = vcf() val format = classOf[SequenceFileInputFormat[Writable, Writable]] - val writables = hadoopFile(path, format, kc.writableClass(km).asInstanceOf[Class[Writable]], - vc.writableClass(vm).asInstanceOf[Class[Writable]], minSplits) + val writables = hadoopFile(path, format, + kc.writableClass(km).asInstanceOf[Class[Writable]], + vc.writableClass(vm).asInstanceOf[Class[Writable]], minSplits) writables.map{case (k,v) => (kc.convert(k), vc.convert(v))} } /** - * Load an RDD saved as a SequenceFile containing serialized objects, with NullWritable keys - * and BytesWritable values that contain a serialized partition. This is still an experimental - * storage format and may not be supported exactly as is in future Spark releases. It will also - * be pretty slow if you use the default serializer (Java serialization), though the nice thing - * about it is that there's very little effort required to save arbitrary objects. + * Load an RDD saved as a SequenceFile containing serialized objects, with NullWritable keys and + * BytesWritable values that contain a serialized partition. This is still an experimental storage + * format and may not be supported exactly as is in future Spark releases. It will also be pretty + * slow if you use the default serializer (Java serialization), though the nice thing about it is + * that there's very little effort required to save arbitrary objects. */ - def objectFile[T: ClassManifest](path: String, minSplits: Int = defaultMinSplits): RDD[T] = { + def objectFile[T: ClassManifest]( + path: String, + minSplits: Int = defaultMinSplits + ): RDD[T] = { sequenceFile(path, classOf[NullWritable], classOf[BytesWritable], minSplits) .flatMap(x => Utils.deserialize[Array[T]](x._2.getBytes)) } /** Build the union of a list of RDDs. */ - def union[T: ClassManifest](rdds: RDD[T]*): RDD[T] = - new UnionRDD(this, rdds) + def union[T: ClassManifest](rdds: RDD[T]*): RDD[T] = new UnionRDD(this, rdds) // Methods for creating shared variables @@ -211,18 +235,17 @@ extends Logging { new Accumulator(initialValue, param) // Keep around a weak hash map of values to Cached versions? - def broadcast[T](value: T) = - Broadcast.getBroadcastFactory.newBroadcast[T] (value, isLocal) + def broadcast[T](value: T) = Broadcast.getBroadcastFactory.newBroadcast[T] (value, isLocal) // Stop the SparkContext def stop() { - scheduler.stop() - scheduler = null - // TODO: Broadcast.stop(), Cache.stop()? - env.mapOutputTracker.stop() - env.cacheTracker.stop() - env.shuffleFetcher.stop() - SparkEnv.set(null) + scheduler.stop() + scheduler = null + // TODO: Broadcast.stop(), Cache.stop()? + env.mapOutputTracker.stop() + env.cacheTracker.stop() + env.shuffleFetcher.stop() + SparkEnv.set(null) } // Wait for the scheduler to be registered @@ -234,25 +257,29 @@ extends Logging { // or the spark.home Java property, or the SPARK_HOME environment variable // (in that order of preference). If neither of these is set, return None. def getSparkHome(): Option[String] = { - if (sparkHome != null) + if (sparkHome != null) { Some(sparkHome) - else if (System.getProperty("spark.home") != null) + } else if (System.getProperty("spark.home") != null) { Some(System.getProperty("spark.home")) - else if (System.getenv("SPARK_HOME") != null) + } else if (System.getenv("SPARK_HOME") != null) { Some(System.getenv("SPARK_HOME")) - else + } else { None + } } /** - * Run a function on a given set of partitions in an RDD and return the results. - * This is the main entry point to the scheduler, by which all actions get launched. - * The allowLocal flag specifies whether the scheduler can run the computation on the - * master rather than shipping it out to the cluster, for short actions like first(). + * Run a function on a given set of partitions in an RDD and return the results. This is the main + * entry point to the scheduler, by which all actions get launched. The allowLocal flag specifies + * whether the scheduler can run the computation on the master rather than shipping it out to the + * cluster, for short actions like first(). */ - def runJob[T, U: ClassManifest](rdd: RDD[T], func: (TaskContext, Iterator[T]) => U, - partitions: Seq[Int], allowLocal: Boolean) - : Array[U] = { + def runJob[T, U: ClassManifest]( + rdd: RDD[T], + func: (TaskContext, Iterator[T]) => U, + partitions: Seq[Int], + allowLocal: Boolean + ): Array[U] = { logInfo("Starting job...") val start = System.nanoTime val result = scheduler.runJob(rdd, func, partitions, allowLocal) @@ -260,22 +287,23 @@ extends Logging { result } - def runJob[T, U: ClassManifest](rdd: RDD[T], func: Iterator[T] => U, partitions: Seq[Int], - allowLocal: Boolean) - : Array[U] = { + def runJob[T, U: ClassManifest]( + rdd: RDD[T], + func: Iterator[T] => U, + partitions: Seq[Int], + allowLocal: Boolean + ): Array[U] = { runJob(rdd, (context: TaskContext, iter: Iterator[T]) => func(iter), partitions, allowLocal) } /** * Run a job on all partitions in an RDD and return the results in an array. */ - def runJob[T, U: ClassManifest](rdd: RDD[T], func: (TaskContext, Iterator[T]) => U) - : Array[U] = { + def runJob[T, U: ClassManifest](rdd: RDD[T], func: (TaskContext, Iterator[T]) => U): Array[U] = { runJob(rdd, func, 0 until rdd.splits.size, false) } - def runJob[T, U: ClassManifest](rdd: RDD[T], func: Iterator[T] => U) - : Array[U] = { + def runJob[T, U: ClassManifest](rdd: RDD[T], func: Iterator[T] => U): Array[U] = { runJob(rdd, func, 0 until rdd.splits.size, false) } @@ -306,10 +334,9 @@ extends Logging { } } - /** - * The SparkContext object contains a number of implicit conversions and - * parameters for use with various Spark features. + * The SparkContext object contains a number of implicit conversions and parameters for use with + * various Spark features. */ object SparkContext { implicit object DoubleAccumulatorParam extends AccumulatorParam[Double] { @@ -323,7 +350,6 @@ object SparkContext { } // TODO: Add AccumulatorParams for other types, e.g. lists and strings - implicit def rddToPairRDDFunctions[K: ClassManifest, V: ClassManifest](rdd: RDD[(K, V)]) = new PairRDDFunctions(rdd) @@ -346,13 +372,14 @@ object SparkContext { implicit def stringToText(s: String) = new Text(s) - private implicit def arrayToArrayWritable[T <% Writable: ClassManifest] (arr: Traversable[T]): ArrayWritable = { + private implicit def arrayToArrayWritable[T <% Writable: ClassManifest](arr: Traversable[T]): ArrayWritable = { def getWritableClass[T <% Writable: ClassManifest](): Class[_ <: Writable] = { val c = { - if (classOf[Writable].isAssignableFrom(classManifest[T].erasure)) + if (classOf[Writable].isAssignableFrom(classManifest[T].erasure)) { classManifest[T].erasure - else + } else { implicitly[T => Writable].getClass.getMethods()(0).getReturnType + } // TODO: use something like WritableConverter to avoid reflection } c.asInstanceOf[Class[ _ <: Writable]] @@ -360,11 +387,11 @@ object SparkContext { def anyToWritable[U <% Writable](u: U): Writable = u - new ArrayWritable(classManifest[T].erasure.asInstanceOf[Class[Writable]], arr.map(x => anyToWritable(x)).toArray) + new ArrayWritable(classManifest[T].erasure.asInstanceOf[Class[Writable]], + arr.map(x => anyToWritable(x)).toArray) } // Helper objects for converting common types to Writable - private def simpleWritableConverter[T, W <: Writable: ClassManifest](convert: W => T) = { val wClass = classManifest[W].erasure.asInstanceOf[Class[W]] new WritableConverter[T](_ => wClass, x => convert(x.asInstanceOf[W])) diff --git a/core/src/main/scala/spark/SparkEnv.scala b/core/src/main/scala/spark/SparkEnv.scala index ad6d54d905..adc7af82c4 100644 --- a/core/src/main/scala/spark/SparkEnv.scala +++ b/core/src/main/scala/spark/SparkEnv.scala @@ -30,8 +30,10 @@ object SparkEnv { val mapOutputTracker = new MapOutputTracker(isMaster) - val shuffleFetcherClass = System.getProperty("spark.shuffle.fetcher", "spark.SimpleShuffleFetcher") - val shuffleFetcher = Class.forName(shuffleFetcherClass).newInstance().asInstanceOf[ShuffleFetcher] + val shuffleFetcherClass = + System.getProperty("spark.shuffle.fetcher", "spark.SimpleShuffleFetcher") + val shuffleFetcher = + Class.forName(shuffleFetcherClass).newInstance().asInstanceOf[ShuffleFetcher] new SparkEnv(cache, serializer, cacheTracker, mapOutputTracker, shuffleFetcher) } diff --git a/core/src/main/scala/spark/Task.scala b/core/src/main/scala/spark/Task.scala index c34083416f..bc3b374344 100644 --- a/core/src/main/scala/spark/Task.scala +++ b/core/src/main/scala/spark/Task.scala @@ -3,7 +3,7 @@ package spark class TaskContext(val stageId: Int, val splitId: Int, val attemptId: Int) extends Serializable abstract class Task[T] extends Serializable { - def run (id: Int): T + def run(id: Int): T def preferredLocations: Seq[String] = Nil def generation: Option[Long] = None } diff --git a/examples/src/main/scala/spark/examples/SparkKMeans.scala b/examples/src/main/scala/spark/examples/SparkKMeans.scala index 3139a0a6e2..a153679ab3 100644 --- a/examples/src/main/scala/spark/examples/SparkKMeans.scala +++ b/examples/src/main/scala/spark/examples/SparkKMeans.scala @@ -54,7 +54,7 @@ object SparkKMeans { while(tempDist > convergeDist) { var closest = data.map (p => (closestPoint(p, kPoints), (p, 1))) - var pointStats = closest.reduceByKey {case ((x1, y1), (x2, y2)) => (x1 + x2, y1+y2)} + var pointStats = closest.reduceByKey {case ((x1, y1), (x2, y2)) => (x1 + x2, y1 + y2)} var newPoints = pointStats.map {pair => (pair._1, pair._2._1 / pair._2._2)}.collect() |