diff --git a/.gitignore b/.gitignore
index 8156d6e8c10ea7ccf6ba0fa7c4c8e76c0c3c4f6a..5abdec5d50d8b68b20a6fbd05bdc1a952ac55370 100644
--- a/.gitignore
+++ b/.gitignore
@@ -7,3 +7,4 @@ third_party/libmesos.so
third_party/libmesos.dylib
conf/java-opts
conf/spark-env.sh
+conf/log4j.properties
diff --git a/Makefile b/Makefile
index 1a3e97a7bea9b9f1353f5bbdf77206637e115f6d..c5d004fb108e785581def5ca35961bd07ad8e2c5 100644
--- a/Makefile
+++ b/Makefile
@@ -12,6 +12,9 @@ JARS += third_party/scalatest-1.2/scalatest-1.2.jar
JARS += third_party/scalacheck_2.8.0-1.7.jar
JARS += third_party/jetty-7.1.6.v20100715/jetty-server-7.1.6.v20100715.jar
JARS += third_party/jetty-7.1.6.v20100715/servlet-api-2.5.jar
+JARS += third_party/apache-log4j-1.2.16/log4j-1.2.16.jar
+JARS += third_party/slf4j-1.6.1/slf4j-api-1.6.1.jar
+JARS += third_party/slf4j-1.6.1/slf4j-log4j12-1.6.1.jar
CLASSPATH = $(subst $(SPACE),:,$(JARS))
SCALA_SOURCES = src/examples/*.scala src/scala/spark/*.scala src/scala/spark/repl/*.scala
diff --git a/conf/log4j.properties b/conf/log4j.properties
new file mode 100644
index 0000000000000000000000000000000000000000..d72dbadc3904f327effddf99594045067be2f529
--- /dev/null
+++ b/conf/log4j.properties
@@ -0,0 +1,8 @@
+# Set everything to be logged to the console
+log4j.rootCategory=INFO, console
+log4j.appender.console=org.apache.log4j.ConsoleAppender
+log4j.appender.console.layout=org.apache.log4j.PatternLayout
+log4j.appender.console.layout.ConversionPattern=%d{yy/MM/dd HH:mm:ss} %p %c{1}: %m%n
+
+# Ignore messages below warning level from Jetty, because it's a bit verbose
+log4j.logger.org.eclipse.jetty=WARN
diff --git a/run b/run
index 33eec9fe1b21dcebe2510c9351c64e3b91cd5f41..f28b39af9b084e84b3a77b51dda3777593694b2a 100755
--- a/run
+++ b/run
@@ -28,21 +28,24 @@ fi
export JAVA_OPTS
# Build up classpath
-SPARK_CLASSPATH="$SPARK_CLASSPATH:$FWDIR/build/classes"
-SPARK_CLASSPATH+=:$FWDIR/third_party/mesos.jar
-SPARK_CLASSPATH+=:$FWDIR/third_party/asm-3.2/lib/all/asm-all-3.2.jar
-SPARK_CLASSPATH+=:$FWDIR/third_party/colt.jar
-SPARK_CLASSPATH+=:$FWDIR/third_party/guava-r06/guava-r06.jar
-SPARK_CLASSPATH+=:$FWDIR/third_party/hadoop-0.20.0/hadoop-0.20.0-core.jar
-SPARK_CLASSPATH+=:third_party/scalatest-1.2/scalatest-1.2.jar
-SPARK_CLASSPATH+=:third_party/scalacheck_2.8.0-1.7.jar
-SPARK_CLASSPATH+=:third_party/jetty-7.1.6.v20100715/jetty-server-7.1.6.v20100715.jar
-SPARK_CLASSPATH+=:third_party/jetty-7.1.6.v20100715/servlet-api-2.5.jar
+CLASSPATH="$SPARK_CLASSPATH:$FWDIR/build/classes"
+CLASSPATH+=:$FWDIR/conf
+CLASSPATH+=:$FWDIR/third_party/mesos.jar
+CLASSPATH+=:$FWDIR/third_party/asm-3.2/lib/all/asm-all-3.2.jar
+CLASSPATH+=:$FWDIR/third_party/colt.jar
+CLASSPATH+=:$FWDIR/third_party/guava-r06/guava-r06.jar
+CLASSPATH+=:$FWDIR/third_party/hadoop-0.20.0/hadoop-0.20.0-core.jar
+CLASSPATH+=:$FWDIR/third_party/scalatest-1.2/scalatest-1.2.jar
+CLASSPATH+=:$FWDIR/third_party/scalacheck_2.8.0-1.7.jar
+CLASSPATH+=:$FWDIR/third_party/jetty-7.1.6.v20100715/jetty-server-7.1.6.v20100715.jar
+CLASSPATH+=:$FWDIR/third_party/jetty-7.1.6.v20100715/servlet-api-2.5.jar
+CLASSPATH+=:$FWDIR/third_party/apache-log4j-1.2.16/log4j-1.2.16.jar
+CLASSPATH+=:$FWDIR/third_party/slf4j-1.6.1/slf4j-api-1.6.1.jar
+CLASSPATH+=:$FWDIR/third_party/slf4j-1.6.1/slf4j-log4j12-1.6.1.jar
for jar in $FWDIR/third_party/hadoop-0.20.0/lib/*.jar; do
- SPARK_CLASSPATH+=:$jar
+ CLASSPATH+=:$jar
done
-export SPARK_CLASSPATH
-export CLASSPATH=$SPARK_CLASSPATH # Needed for spark-shell
+export CLASSPATH # Needed for spark-shell
if [ -n "$SCALA_HOME" ]; then
SCALA=${SCALA_HOME}/bin/scala
@@ -50,4 +53,4 @@ else
SCALA=scala
fi
-exec $SCALA -cp $SPARK_CLASSPATH $@
+exec $SCALA -cp $CLASSPATH $@
diff --git a/spark-executor b/spark-executor
index ee847cfff03ebb9a78017f078a287ecc4341949c..0f9b9b1ece1a27bef97be6fe9c705a5ee9e6632f 100755
--- a/spark-executor
+++ b/spark-executor
@@ -1,5 +1,4 @@
#!/bin/sh
-echo "In spark-executor"
FWDIR="`dirname $0`"
-echo Framework dir: $FWDIR
+echo "Running spark-executor with framework dir = $FWDIR"
exec $FWDIR/run spark.Executor
diff --git a/src/scala/spark/Broadcast.scala b/src/scala/spark/Broadcast.scala
index 7fe84da47c94a9c82ff6ad5267d0a2b4cebe859d..fc376359838a2fca60c65d94b58355ee548b82fe 100644
--- a/src/scala/spark/Broadcast.scala
+++ b/src/scala/spark/Broadcast.scala
@@ -33,7 +33,7 @@ trait BroadcastRecipe {
// TODO: Right, now no parallelization between multiple broadcasts
@serializable
class ChainedStreamingBroadcast[T] (@transient var value_ : T, local: Boolean)
- extends BroadcastRecipe {
+extends BroadcastRecipe with Logging {
def value = value_
@@ -92,7 +92,7 @@ class ChainedStreamingBroadcast[T] (@transient var value_ : T, local: Boolean)
}
val time = (System.nanoTime - start) / 1e9
- println( System.currentTimeMillis + ": " + "Reading Broadcasted variable " + uuid + " took " + time + " s")
+ logInfo("Reading Broadcasted variable " + uuid + " took " + time + " s")
}
}
}
@@ -149,7 +149,7 @@ class ChainedStreamingBroadcast[T] (@transient var value_ : T, local: Boolean)
@serializable
class CentralizedHDFSBroadcast[T](@transient var value_ : T, local: Boolean)
- extends BroadcastRecipe {
+extends BroadcastRecipe with Logging {
def value = value_
@@ -179,7 +179,7 @@ class CentralizedHDFSBroadcast[T](@transient var value_ : T, local: Boolean)
fileIn.close
val time = (System.nanoTime - start) / 1e9
- println( System.currentTimeMillis + ": " + "Reading Broadcasted variable " + uuid + " took " + time + " s")
+ logInfo("Reading Broadcasted variable " + uuid + " took " + time + " s")
}
}
}
@@ -188,7 +188,7 @@ class CentralizedHDFSBroadcast[T](@transient var value_ : T, local: Boolean)
@serializable
case class SourceInfo (val hostAddress: String, val listenPort: Int,
val totalBlocks: Int, val totalBytes: Int, val replicaID: Int)
- extends Comparable [SourceInfo]{
+extends Comparable[SourceInfo]{
var currentLeechers = 0
var receptionFailed = false
@@ -231,7 +231,7 @@ private object Broadcast {
}
}
-private object BroadcastCS {
+private object BroadcastCS extends Logging {
val values = new MapMaker ().softValues ().makeMap[UUID, Any]
// val valueInfos = new MapMaker ().softValues ().makeMap[UUID, Any]
@@ -286,15 +286,15 @@ private object BroadcastCS {
guideMR = new GuideMultipleRequests
guideMR.setDaemon (true)
guideMR.start
- println (System.currentTimeMillis + ": " + "GuideMultipleRequests started")
+ logInfo("GuideMultipleRequests started")
}
serveMR = new ServeMultipleRequests
serveMR.setDaemon (true)
serveMR.start
- println (System.currentTimeMillis + ": " + "ServeMultipleRequests started")
+ logInfo("ServeMultipleRequests started")
- println (System.currentTimeMillis + ": " + "BroadcastCS object has been initialized")
+ logInfo("BroadcastCS object has been initialized")
initialized = true
}
@@ -352,7 +352,7 @@ private object BroadcastCS {
// Connect to Master and send this worker's Information
val clientSocketToMaster =
new Socket(BroadcastCS.masterHostAddress, BroadcastCS.masterListenPort)
- println (System.currentTimeMillis + ": " + "Connected to Master's guiding object")
+ logInfo("Connected to Master's guiding object")
// TODO: Guiding object connection is reusable
val oisMaster =
new ObjectInputStream (clientSocketToMaster.getInputStream)
@@ -371,11 +371,11 @@ private object BroadcastCS {
}
totalBytes = sourceInfo.totalBytes
- println (System.currentTimeMillis + ": " + "Received SourceInfo from Master:" + sourceInfo + " My Port: " + listenPort)
+ logInfo("Received SourceInfo from Master:" + sourceInfo + " My Port: " + listenPort)
retByteArray = receiveSingleTransmission (sourceInfo)
- println (System.currentTimeMillis + ": " + "I got this from receiveSingleTransmission: " + retByteArray)
+ logInfo("I got this from receiveSingleTransmission: " + retByteArray)
// TODO: Update sourceInfo to add error notifactions for Master
if (retByteArray == null) { sourceInfo.receptionFailed = true }
@@ -414,8 +414,8 @@ private object BroadcastCS {
oisSource =
new ObjectInputStream (clientSocketToSource.getInputStream)
- println (System.currentTimeMillis + ": " + "Inside receiveSingleTransmission")
- println (System.currentTimeMillis + ": " + "totalBlocks: "+ totalBlocks + " " + "hasBlocks: " + hasBlocks)
+ logInfo("Inside receiveSingleTransmission")
+ logInfo("totalBlocks: "+ totalBlocks + " " + "hasBlocks: " + hasBlocks)
retByteArray = new Array[Byte] (totalBytes)
for (i <- 0 until totalBlocks) {
val bcBlock = oisSource.readObject.asInstanceOf[BroadcastBlock]
@@ -426,14 +426,14 @@ private object BroadcastCS {
hasBlocksLock.synchronized {
hasBlocksLock.notifyAll
}
- println (System.currentTimeMillis + ": " + "Received block: " + i + " " + bcBlock)
+ logInfo("Received block: " + i + " " + bcBlock)
}
assert (hasBlocks == totalBlocks)
- println (System.currentTimeMillis + ": " + "After the receive loop")
+ logInfo("After the receive loop")
} catch {
case e: Exception => {
retByteArray = null
- println (System.currentTimeMillis + ": " + "receiveSingleTransmission had a " + e)
+ logInfo("receiveSingleTransmission had a " + e)
}
} finally {
if (oisSource != null) { oisSource.close }
@@ -446,13 +446,13 @@ private object BroadcastCS {
return retByteArray
}
- class TrackMultipleValues extends Thread {
+ class TrackMultipleValues extends Thread with Logging {
override def run = {
var threadPool = Executors.newCachedThreadPool
var serverSocket: ServerSocket = null
serverSocket = new ServerSocket (BroadcastCS.masterListenPort)
- println (System.currentTimeMillis + ": " + "TrackMultipleVariables" + serverSocket + " " + listenPort)
+ logInfo("TrackMultipleVariables" + serverSocket + " " + listenPort)
var keepAccepting = true
try {
@@ -463,11 +463,11 @@ private object BroadcastCS {
clientSocket = serverSocket.accept
} catch {
case e: Exception => {
- println ("TrackMultipleValues Timeout. Stopping listening...")
+ logInfo("TrackMultipleValues Timeout. Stopping listening...")
keepAccepting = false
}
}
- println (System.currentTimeMillis + ": " + "TrackMultipleValues:Got new request:" + clientSocket)
+ logInfo("TrackMultipleValues:Got new request:" + clientSocket)
if (clientSocket != null) {
try {
threadPool.execute (new Runnable {
@@ -506,14 +506,14 @@ private object BroadcastCS {
}
- class GuideMultipleRequests extends Thread {
+ class GuideMultipleRequests extends Thread with Logging {
override def run = {
var threadPool = Executors.newCachedThreadPool
var serverSocket: ServerSocket = null
serverSocket = new ServerSocket (BroadcastCS.masterListenPort)
// listenPort = BroadcastCS.masterListenPort
- println (System.currentTimeMillis + ": " + "GuideMultipleRequests" + serverSocket + " " + listenPort)
+ logInfo("GuideMultipleRequests" + serverSocket + " " + listenPort)
var keepAccepting = true
try {
@@ -524,12 +524,12 @@ private object BroadcastCS {
clientSocket = serverSocket.accept
} catch {
case e: Exception => {
- println ("GuideMultipleRequests Timeout. Stopping listening...")
+ logInfo("GuideMultipleRequests Timeout. Stopping listening...")
keepAccepting = false
}
}
if (clientSocket != null) {
- println (System.currentTimeMillis + ": " + "Guide:Accepted new client connection:" + clientSocket)
+ logInfo("Guide:Accepted new client connection:" + clientSocket)
try {
threadPool.execute (new GuideSingleRequest (clientSocket))
} catch {
@@ -543,7 +543,8 @@ private object BroadcastCS {
}
}
- class GuideSingleRequest (val clientSocket: Socket) extends Runnable {
+ class GuideSingleRequest (val clientSocket: Socket)
+ extends Runnable with Logging {
private val oos = new ObjectOutputStream (clientSocket.getOutputStream)
private val ois = new ObjectInputStream (clientSocket.getInputStream)
@@ -552,21 +553,21 @@ private object BroadcastCS {
def run = {
try {
- println (System.currentTimeMillis + ": " + "new GuideSingleRequest is running")
+ logInfo("new GuideSingleRequest is running")
// Connecting worker is sending in its hostAddress and listenPort it will
// be listening to. ReplicaID is 0 and other fields are invalid (-1)
var sourceInfo = ois.readObject.asInstanceOf[SourceInfo]
// Select a suitable source and send it back to the worker
selectedSourceInfo = selectSuitableSource (sourceInfo)
- println (System.currentTimeMillis + ": " + "Sending selectedSourceInfo:" + selectedSourceInfo)
+ logInfo("Sending selectedSourceInfo:" + selectedSourceInfo)
oos.writeObject (selectedSourceInfo)
oos.flush
// Add this new (if it can finish) source to the PQ of sources
thisWorkerInfo = new SourceInfo(sourceInfo.hostAddress,
sourceInfo.listenPort, totalBlocks, totalBytes, 0)
- println (System.currentTimeMillis + ": " + "Adding possible new source to pqOfSources: " + thisWorkerInfo)
+ logInfo("Adding possible new source to pqOfSources: " + thisWorkerInfo)
pqOfSources.synchronized {
pqOfSources.add (thisWorkerInfo)
}
@@ -642,14 +643,14 @@ private object BroadcastCS {
}
}
- class ServeMultipleRequests extends Thread {
+ class ServeMultipleRequests extends Thread with Logging {
override def run = {
var threadPool = Executors.newCachedThreadPool
var serverSocket: ServerSocket = null
serverSocket = new ServerSocket (0)
listenPort = serverSocket.getLocalPort
- println (System.currentTimeMillis + ": " + "ServeMultipleRequests" + serverSocket + " " + listenPort)
+ logInfo("ServeMultipleRequests" + serverSocket + " " + listenPort)
listenPortLock.synchronized {
listenPortLock.notifyAll
@@ -664,12 +665,12 @@ private object BroadcastCS {
clientSocket = serverSocket.accept
} catch {
case e: Exception => {
- println ("ServeMultipleRequests Timeout. Stopping listening...")
+ logInfo("ServeMultipleRequests Timeout. Stopping listening...")
keepAccepting = false
}
}
if (clientSocket != null) {
- println (System.currentTimeMillis + ": " + "Serve:Accepted new client connection:" + clientSocket)
+ logInfo("Serve:Accepted new client connection:" + clientSocket)
try {
threadPool.execute (new ServeSingleRequest (clientSocket))
} catch {
@@ -683,23 +684,24 @@ private object BroadcastCS {
}
}
- class ServeSingleRequest (val clientSocket: Socket) extends Runnable {
+ class ServeSingleRequest (val clientSocket: Socket)
+ extends Runnable with Logging {
private val oos = new ObjectOutputStream (clientSocket.getOutputStream)
private val ois = new ObjectInputStream (clientSocket.getInputStream)
def run = {
try {
- println (System.currentTimeMillis + ": " + "new ServeSingleRequest is running")
+ logInfo("new ServeSingleRequest is running")
sendObject
} catch {
// TODO: Need to add better exception handling here
// If something went wrong, e.g., the worker at the other end died etc.
// then close everything up
case e: Exception => {
- println (System.currentTimeMillis + ": " + "ServeSingleRequest had a " + e)
+ logInfo("ServeSingleRequest had a " + e)
}
} finally {
- println (System.currentTimeMillis + ": " + "ServeSingleRequest is closing streams and sockets")
+ logInfo("ServeSingleRequest is closing streams and sockets")
ois.close
oos.close
clientSocket.close
@@ -726,7 +728,7 @@ private object BroadcastCS {
} catch {
case e: Exception => { }
}
- println (System.currentTimeMillis + ": " + "Send block: " + i + " " + arrayOfBlocks(i))
+ logInfo("Send block: " + i + " " + arrayOfBlocks(i))
}
}
}
@@ -734,7 +736,7 @@ private object BroadcastCS {
}
}
-private object BroadcastCH {
+private object BroadcastCH extends Logging {
val values = new MapMaker ().softValues ().makeMap[UUID, Any]
private var initialized = false
diff --git a/src/scala/spark/ClosureCleaner.scala b/src/scala/spark/ClosureCleaner.scala
index 8037434c387ad0777774519fb33471e989f6813b..0e0b3954d4db9703da58957b19b5d0af18f3e6dc 100644
--- a/src/scala/spark/ClosureCleaner.scala
+++ b/src/scala/spark/ClosureCleaner.scala
@@ -8,7 +8,7 @@ import org.objectweb.asm.commons.EmptyVisitor
import org.objectweb.asm.Opcodes._
-object ClosureCleaner {
+object ClosureCleaner extends Logging {
private def getClassReader(cls: Class[_]): ClassReader = {
new ClassReader(cls.getResourceAsStream(
cls.getName.replaceFirst("^.*\\.", "") + ".class"))
@@ -72,13 +72,13 @@ object ClosureCleaner {
val field = cls.getDeclaredField(fieldName)
field.setAccessible(true)
val value = field.get(obj)
- //println("1: Setting " + fieldName + " on " + cls + " to " + value);
+ //logInfo("1: Setting " + fieldName + " on " + cls + " to " + value);
field.set(outer, value)
}
}
if (outer != null) {
- //println("2: Setting $outer on " + func.getClass + " to " + outer);
+ //logInfo("2: Setting $outer on " + func.getClass + " to " + outer);
val field = func.getClass.getDeclaredField("$outer")
field.setAccessible(true)
field.set(func, outer)
@@ -101,7 +101,7 @@ object ClosureCleaner {
val newCtor = rf.newConstructorForSerialization(cls, parentCtor)
val obj = newCtor.newInstance().asInstanceOf[AnyRef];
if (outer != null) {
- //println("3: Setting $outer on " + cls + " to " + outer);
+ //logInfo("3: Setting $outer on " + cls + " to " + outer);
val field = cls.getDeclaredField("$outer")
field.setAccessible(true)
field.set(obj, outer)
diff --git a/src/scala/spark/Executor.scala b/src/scala/spark/Executor.scala
index 58b20b41dc1fde1ec4200b534a562387edba8540..be73aae541d6822f2fdea62350aa09b1ee4e1a5f 100644
--- a/src/scala/spark/Executor.scala
+++ b/src/scala/spark/Executor.scala
@@ -5,10 +5,14 @@ import java.util.concurrent.{Executors, ExecutorService}
import mesos.{ExecutorArgs, ExecutorDriver, MesosExecutorDriver}
import mesos.{TaskDescription, TaskState, TaskStatus}
-object Executor {
+/**
+ * The Mesos executor for Spark.
+ */
+object Executor extends Logging {
def main(args: Array[String]) {
System.loadLibrary("mesos")
+ // Create a new Executor implementation that will run our tasks
val exec = new mesos.Executor() {
var classLoader: ClassLoader = null
var threadPool: ExecutorService = null
@@ -27,7 +31,7 @@ object Executor {
classLoader = this.getClass.getClassLoader
val classUri = System.getProperty("spark.repl.class.uri")
if (classUri != null) {
- println("Using REPL class URI: " + classUri)
+ logInfo("Using REPL class URI: " + classUri)
classLoader = new repl.ExecutorClassLoader(classUri, classLoader)
}
Thread.currentThread.setContextClassLoader(classLoader)
@@ -43,7 +47,7 @@ object Executor {
val arg = desc.getArg
threadPool.execute(new Runnable() {
def run() = {
- println("Running task ID " + taskId)
+ logInfo("Running task ID " + taskId)
try {
Accumulators.clear
val task = Utils.deserialize[Task[Any]](arg, classLoader)
@@ -52,12 +56,11 @@ object Executor {
val result = new TaskResult(value, accumUpdates)
d.sendStatusUpdate(new TaskStatus(
taskId, TaskState.TASK_FINISHED, Utils.serialize(result)))
- println("Finished task ID " + taskId)
+ logInfo("Finished task ID " + taskId)
} catch {
case e: Exception => {
// TODO: Handle errors in tasks less dramatically
- System.err.println("Exception in task ID " + taskId + ":")
- e.printStackTrace
+ logError("Exception in task ID " + taskId, e)
System.exit(1)
}
}
@@ -66,6 +69,7 @@ object Executor {
}
}
+ // Start it running and connect it to the slave
new MesosExecutorDriver(exec).run()
}
}
diff --git a/src/scala/spark/LocalScheduler.scala b/src/scala/spark/LocalScheduler.scala
index a3e0fae0653a2bf068b96f968d3c118e4d1ab523..20954a1224243b68565cdbb63d420656ab345dbd 100644
--- a/src/scala/spark/LocalScheduler.scala
+++ b/src/scala/spark/LocalScheduler.scala
@@ -4,8 +4,10 @@ import java.util.concurrent._
import scala.collection.mutable.Map
-// A simple Scheduler implementation that runs tasks locally in a thread pool.
-private class LocalScheduler(threads: Int) extends Scheduler {
+/**
+ * A simple Scheduler implementation that runs tasks locally in a thread pool.
+ */
+private class LocalScheduler(threads: Int) extends Scheduler with Logging {
var threadPool: ExecutorService =
Executors.newFixedThreadPool(threads, DaemonThreadFactory)
@@ -20,25 +22,24 @@ private class LocalScheduler(threads: Int) extends Scheduler {
for (i <- 0 until tasks.length) {
futures(i) = threadPool.submit(new Callable[TaskResult[T]]() {
def call(): TaskResult[T] = {
- println("Running task " + i)
+ logInfo("Running task " + i)
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 Nexus Executor works
Accumulators.clear
val bytes = Utils.serialize(tasks(i))
- println("Size of task " + i + " is " + bytes.size + " bytes")
+ logInfo("Size of task " + i + " is " + bytes.size + " bytes")
val task = Utils.deserialize[Task[T]](
bytes, currentThread.getContextClassLoader)
val value = task.run
val accumUpdates = Accumulators.values
- println("Finished task " + i)
+ logInfo("Finished task " + i)
new TaskResult[T](value, accumUpdates)
} catch {
case e: Exception => {
// TODO: Do something nicer here
- System.err.println("Exception in task " + i + ":")
- e.printStackTrace()
+ logError("Exception in task " + i, e)
System.exit(1)
null
}
@@ -58,7 +59,10 @@ private class LocalScheduler(threads: Int) extends Scheduler {
override def numCores() = threads
}
-// A ThreadFactory that creates daemon threads
+
+/**
+ * A ThreadFactory that creates daemon threads
+ */
private object DaemonThreadFactory extends ThreadFactory {
override def newThread(r: Runnable): Thread = {
val t = new Thread(r);
diff --git a/src/scala/spark/Logging.scala b/src/scala/spark/Logging.scala
new file mode 100644
index 0000000000000000000000000000000000000000..2d1feebbb1ede64fe2899d6a9608b6477e965991
--- /dev/null
+++ b/src/scala/spark/Logging.scala
@@ -0,0 +1,49 @@
+package spark
+
+import org.slf4j.Logger
+import org.slf4j.LoggerFactory
+
+/**
+ * Utility trait for classes that want to log data. Creates a SLF4J logger
+ * for the class and allows logging messages at different levels using
+ * methods that only evaluate parameters lazily if the log level is enabled.
+ */
+trait Logging {
+ // Make the log field transient so that objects with Logging can
+ // be serialized and used on another machine
+ @transient private var log_ : Logger = null
+
+ // Method to get or create the logger for this object
+ def log: Logger = {
+ if (log_ == null) {
+ var className = this.getClass().getName()
+ // Ignore trailing $'s in the class names for Scala objects
+ if (className.endsWith("$"))
+ className = className.substring(0, className.length - 1)
+ log_ = LoggerFactory.getLogger(className)
+ }
+ return log_
+ }
+
+ // Log methods that take only a String
+ def logInfo(msg: => String) = if (log.isInfoEnabled) log.info(msg)
+
+ def logDebug(msg: => String) = if (log.isDebugEnabled) log.debug(msg)
+
+ def logWarning(msg: => String) = if (log.isWarnEnabled) log.warn(msg)
+
+ def logError(msg: => String) = if (log.isErrorEnabled) log.error(msg)
+
+ // Log methods that take Throwables (Exceptions/Errors) too
+ def logInfo(msg: => String, throwable: Throwable) =
+ if (log.isInfoEnabled) log.info(msg)
+
+ def logDebug(msg: => String, throwable: Throwable) =
+ if (log.isDebugEnabled) log.debug(msg)
+
+ def logWarning(msg: => String, throwable: Throwable) =
+ if (log.isWarnEnabled) log.warn(msg, throwable)
+
+ def logError(msg: => String, throwable: Throwable) =
+ if (log.isErrorEnabled) log.error(msg, throwable)
+}
diff --git a/src/scala/spark/MesosScheduler.scala b/src/scala/spark/MesosScheduler.scala
index 2face7f08d3fcc756d2ef6503be2ca0f0efd7f0c..873a97c59c14436bad326ac985c746e5b1750508 100644
--- a/src/scala/spark/MesosScheduler.scala
+++ b/src/scala/spark/MesosScheduler.scala
@@ -23,7 +23,7 @@ import mesos._
// all the offers to the ParallelOperation and have it load-balance.
private class MesosScheduler(
master: String, frameworkName: String, execArg: Array[Byte])
-extends NScheduler with spark.Scheduler
+extends NScheduler with spark.Scheduler with Logging
{
// Lock used by runTasks to ensure only one thread can be in it
val runTasksMutex = new Object()
@@ -101,7 +101,7 @@ extends NScheduler with spark.Scheduler
}
override def registered(d: SchedulerDriver, frameworkId: String) {
- println("Registered as framework ID " + frameworkId)
+ logInfo("Registered as framework ID " + frameworkId)
registeredLock.synchronized {
isRegistered = true
registeredLock.notifyAll()
@@ -137,7 +137,7 @@ extends NScheduler with spark.Scheduler
case None => {}
}
} catch {
- case e: Exception => e.printStackTrace
+ case e: Exception => logError("Exception in resourceOffer", e)
}
}
}
@@ -157,11 +157,11 @@ extends NScheduler with spark.Scheduler
activeOps(opId).statusUpdate(status)
}
case None =>
- println("TID " + status.getTaskId + " already finished")
+ logInfo("TID " + status.getTaskId + " already finished")
}
} catch {
- case e: Exception => e.printStackTrace
+ case e: Exception => logError("Exception in statusUpdate", e)
}
}
}
@@ -173,12 +173,11 @@ extends NScheduler with spark.Scheduler
try {
activeOp.error(code, message)
} catch {
- case e: Exception => e.printStackTrace
+ case e: Exception => logError("Exception in error callback", e)
}
}
} else {
- val msg = "Mesos error: %s (error code: %d)".format(message, code)
- System.err.println(msg)
+ logError("Mesos error: %s (error code: %d)".format(message, code))
System.exit(1)
}
}
@@ -205,7 +204,7 @@ trait ParallelOperation {
class SimpleParallelOperation[T: ClassManifest](
sched: MesosScheduler, tasks: Array[Task[T]], val opId: Int)
-extends ParallelOperation
+extends ParallelOperation with Logging
{
// Maximum time to wait to run a task in a preferred location (in ms)
val LOCALITY_WAIT = System.getProperty("spark.locality.wait", "3000").toLong
@@ -261,9 +260,11 @@ extends ParallelOperation
val taskId = sched.newTaskId()
sched.taskIdToOpId(taskId) = opId
tidToIndex(taskId) = i
- printf("Starting task %d as opId %d, TID %s on slave %s: %s (%s)",
- i, opId, taskId, offer.getSlaveId, offer.getHost,
- if(checkPref) "preferred" else "non-preferred")
+ val preferred = if(checkPref) "preferred" else "non-preferred"
+ val message =
+ "Starting task %d as opId %d, TID %s on slave %s: %s (%s)".format(
+ i, opId, taskId, offer.getSlaveId, offer.getHost, preferred)
+ logInfo(message)
tasks(i).markStarted(offer)
launched(i) = true
tasksLaunched += 1
@@ -273,7 +274,7 @@ extends ParallelOperation
params.put("cpus", "" + desiredCpus)
params.put("mem", "" + desiredMem)
val serializedTask = Utils.serialize(tasks(i))
- println("... Serialized size: " + serializedTask.size)
+ //logInfo("Serialized size: " + serializedTask.size)
return Some(new TaskDescription(taskId, offer.getSlaveId,
"task_" + taskId, params, serializedTask))
}
@@ -298,36 +299,39 @@ extends ParallelOperation
def taskFinished(status: TaskStatus) {
val tid = status.getTaskId
- print("Finished opId " + opId + " TID " + tid)
- if (!finished(tidToIndex(tid))) {
+ val index = tidToIndex(tid)
+ if (!finished(index)) {
+ tasksFinished += 1
+ logInfo("Finished opId %d TID %d (progress: %d/%d)".format(
+ opId, tid, tasksFinished, numTasks))
// Deserialize task result
val result = Utils.deserialize[TaskResult[T]](status.getData)
- results(tidToIndex(tid)) = result.value
+ results(index) = result.value
// Update accumulators
Accumulators.add(callingThread, result.accumUpdates)
// Mark finished and stop if we've finished all the tasks
- finished(tidToIndex(tid)) = true
+ finished(index) = true
// Remove TID -> opId mapping from sched
sched.taskIdToOpId.remove(tid)
- tasksFinished += 1
-
- println(", finished " + tasksFinished + "/" + numTasks)
if (tasksFinished == numTasks)
setAllFinished()
} else {
- printf("... Task %s had already finished, so ignoring it\n", tidToIndex(tid))
+ logInfo("Ignoring task-finished event for TID " + tid +
+ " because task " + index + " is already finished")
}
}
def taskLost(status: TaskStatus) {
val tid = status.getTaskId
- println("Lost opId " + opId + " TID " + tid)
- if (!finished(tidToIndex(tid))) {
- launched(tidToIndex(tid)) = false
+ val index = tidToIndex(tid)
+ if (!finished(index)) {
+ logInfo("Lost opId " + opId + " TID " + tid)
+ launched(index) = false
sched.taskIdToOpId.remove(tid)
tasksLaunched -= 1
} else {
- printf("Task %s had already finished, so ignoring it\n", tidToIndex(tid))
+ logInfo("Ignoring task-lost event for TID " + tid +
+ " because task " + index + " is already finished")
}
}
diff --git a/src/scala/spark/RDD.scala b/src/scala/spark/RDD.scala
index f8d1ab7f570cfade40d53f891c0911f5d8fca664..4d0c8c671160763127f7fab9a2f52a6f2bb2d712 100644
--- a/src/scala/spark/RDD.scala
+++ b/src/scala/spark/RDD.scala
@@ -93,27 +93,27 @@ extends Task[U] {
class ForeachTask[T: ClassManifest](
rdd: RDD[T], split: Split, func: T => Unit)
-extends RDDTask[Unit, T](rdd, split) {
+extends RDDTask[Unit, T](rdd, split) with Logging {
override def run() {
- println("Processing " + split)
+ logInfo("Processing " + split)
rdd.iterator(split).foreach(func)
}
}
class CollectTask[T](
rdd: RDD[T], split: Split)(implicit m: ClassManifest[T])
-extends RDDTask[Array[T], T](rdd, split) {
+extends RDDTask[Array[T], T](rdd, split) with Logging {
override def run(): Array[T] = {
- println("Processing " + split)
+ logInfo("Processing " + split)
rdd.iterator(split).toArray(m)
}
}
class ReduceTask[T: ClassManifest](
rdd: RDD[T], split: Split, f: (T, T) => T)
-extends RDDTask[Option[T], T](rdd, split) {
+extends RDDTask[Option[T], T](rdd, split) with Logging {
override def run(): Option[T] = {
- println("Processing " + split)
+ logInfo("Processing " + split)
val iter = rdd.iterator(split)
if (iter.hasNext)
Some(iter.reduceLeft(f))
@@ -183,7 +183,7 @@ extends RDD[T](prev.sparkContext) {
class CachedRDD[T](
prev: RDD[T])(implicit m: ClassManifest[T])
-extends RDD[T](prev.sparkContext) {
+extends RDD[T](prev.sparkContext) with Logging {
val id = CachedRDD.newId()
@transient val cacheLocs = Map[Split, List[String]]()
@@ -217,7 +217,7 @@ extends RDD[T](prev.sparkContext) {
}
}
// If we got here, we have to load the split
- println("Loading and caching " + split)
+ logInfo("Loading and caching " + split)
val array = prev.iterator(split).toArray(m)
cache.put(key, array)
loading.synchronized {
diff --git a/src/scala/spark/SparkContext.scala b/src/scala/spark/SparkContext.scala
index d5d4db4678eb4df00df59350286eea879d52e276..e00eda9aa24528e78e62578a1626d04f8c4f33aa 100644
--- a/src/scala/spark/SparkContext.scala
+++ b/src/scala/spark/SparkContext.scala
@@ -6,7 +6,7 @@ import java.util.UUID
import scala.collection.mutable.ArrayBuffer
import scala.actors.Actor._
-class SparkContext(master: String, frameworkName: String) {
+class SparkContext(master: String, frameworkName: String) extends Logging {
Broadcast.initialize(true)
def parallelize[T: ClassManifest](seq: Seq[T], numSlices: Int) =
@@ -56,10 +56,10 @@ class SparkContext(master: String, frameworkName: String) {
private[spark] def runTaskObjects[T: ClassManifest](tasks: Seq[Task[T]])
: Array[T] = {
- println("Running " + tasks.length + " tasks in parallel")
+ logInfo("Running " + tasks.length + " tasks in parallel")
val start = System.nanoTime
val result = scheduler.runTasks(tasks.toArray)
- println("Tasks finished in " + (System.nanoTime - start) / 1e9 + " s")
+ logInfo("Tasks finished in " + (System.nanoTime - start) / 1e9 + " s")
return result
}
diff --git a/src/scala/spark/repl/ClassServer.scala b/src/scala/spark/repl/ClassServer.scala
index 14ab2fe2a30ef2d827ea08a2d0e0599fdbfde026..6a40d92765b922cd9ca72fd9d91e13a7667288ad 100644
--- a/src/scala/spark/repl/ClassServer.scala
+++ b/src/scala/spark/repl/ClassServer.scala
@@ -8,6 +8,8 @@ import org.eclipse.jetty.server.handler.DefaultHandler
import org.eclipse.jetty.server.handler.HandlerList
import org.eclipse.jetty.server.handler.ResourceHandler
+import spark.Logging
+
/**
* Exception type thrown by ClassServer when it is in the wrong state
@@ -21,7 +23,7 @@ class ServerStateException(message: String) extends Exception(message)
* class files created as the user types in lines of code. This is just a
* wrapper around a Jetty embedded HTTP server.
*/
-class ClassServer(classDir: File) {
+class ClassServer(classDir: File) extends Logging {
private var server: Server = null
private var port: Int = -1
@@ -37,6 +39,7 @@ class ClassServer(classDir: File) {
server.setHandler(handlerList)
server.start()
port = server.getConnectors()(0).getLocalPort()
+ logDebug("ClassServer started at " + uri)
}
}
diff --git a/src/scala/spark/repl/ExecutorClassLoader.scala b/src/scala/spark/repl/ExecutorClassLoader.scala
index 0833b319f7f3d564713d91938e094ccf01e6a9d0..13d81ec1cf096befaff8b7e9901b783faa17212e 100644
--- a/src/scala/spark/repl/ExecutorClassLoader.scala
+++ b/src/scala/spark/repl/ExecutorClassLoader.scala
@@ -31,7 +31,6 @@ extends ClassLoader(parent) {
override def findClass(name: String): Class[_] = {
try {
- //println("repl.ExecutorClassLoader resolving " + name)
val pathInDirectory = name.replace('.', '/') + ".class"
val inputStream = {
if (fileSystem != null)
@@ -92,7 +91,6 @@ extends ClassAdapter(cv) {
// This is the constructor, time to clean it; just output some new
// instructions to mv that create the object and set the static MODULE$
// field in the class to point to it, but do nothing otherwise.
- //println("Cleaning constructor of " + className)
mv.visitCode()
mv.visitVarInsn(ALOAD, 0) // load this
mv.visitMethodInsn(INVOKESPECIAL, "java/lang/Object", "<init>", "()V")
diff --git a/src/scala/spark/repl/SparkInterpreter.scala b/src/scala/spark/repl/SparkInterpreter.scala
index 29a442095020dec8133c6a050493df2c1622b4dd..ae2e7e8a681a5d9c466c8e0db804ad9c1052aa6d 100644
--- a/src/scala/spark/repl/SparkInterpreter.scala
+++ b/src/scala/spark/repl/SparkInterpreter.scala
@@ -90,6 +90,7 @@ class SparkInterpreter(val settings: Settings, out: PrintWriter) {
val SPARK_DEBUG_REPL: Boolean = (System.getenv("SPARK_DEBUG_REPL") == "1")
+ /** Local directory to save .class files too */
val outputDir = {
val rootDir = new File(System.getProperty("spark.repl.classdir",
System.getProperty("java.io.tmpdir")))
@@ -108,22 +109,26 @@ class SparkInterpreter(val settings: Settings, out: PrintWriter) {
dir = null
} catch { case e: IOException => ; }
}
- if (SPARK_DEBUG_REPL)
+ if (SPARK_DEBUG_REPL) {
println("Output directory: " + dir)
+ }
dir
}
- /** directory to save .class files to */
+ /** Scala compiler virtual directory for outputDir */
//val virtualDirectory = new VirtualDirectory("(memory)", None)
val virtualDirectory = new PlainFile(outputDir)
/** Jetty server that will serve our classes to worker nodes */
val classServer = new ClassServer(outputDir)
- // Start the classServer and remember its URI in a spark system property */
+ // Start the classServer and store its URI in a spark system property
+ // (which will be passed to executors so that they can connect to it)
classServer.start()
- println("ClassServer started, URI = " + classServer.uri)
System.setProperty("spark.repl.class.uri", classServer.uri)
+ if (SPARK_DEBUG_REPL) {
+ println("ClassServer started, URI = " + classServer.uri)
+ }
/** reporter */
object reporter extends ConsoleReporter(settings, null, out) {
diff --git a/third_party/apache-log4j-1.2.16/log4j-1.2.16.jar b/third_party/apache-log4j-1.2.16/log4j-1.2.16.jar
new file mode 100644
index 0000000000000000000000000000000000000000..3f9d847618bcec9c81f36ad6ee76a0f53816987f
Binary files /dev/null and b/third_party/apache-log4j-1.2.16/log4j-1.2.16.jar differ
diff --git a/third_party/hadoop-0.20.0/lib/slf4j-api-1.4.3.jar b/third_party/hadoop-0.20.0/lib/slf4j-api-1.4.3.jar
deleted file mode 100644
index b34fe8d82de073b1c14316c3228bb221fa623ae2..0000000000000000000000000000000000000000
Binary files a/third_party/hadoop-0.20.0/lib/slf4j-api-1.4.3.jar and /dev/null differ
diff --git a/third_party/hadoop-0.20.0/lib/slf4j-log4j12-1.4.3.jar b/third_party/hadoop-0.20.0/lib/slf4j-log4j12-1.4.3.jar
deleted file mode 100644
index 70082fc382bed73c3eb699b988e2ea1de0c95aac..0000000000000000000000000000000000000000
Binary files a/third_party/hadoop-0.20.0/lib/slf4j-log4j12-1.4.3.jar and /dev/null differ
diff --git a/third_party/slf4j-1.6.1/slf4j-api-1.6.1.jar b/third_party/slf4j-1.6.1/slf4j-api-1.6.1.jar
new file mode 100644
index 0000000000000000000000000000000000000000..42e0ad0de7773da9b94b12f503deda7f5a506015
Binary files /dev/null and b/third_party/slf4j-1.6.1/slf4j-api-1.6.1.jar differ
diff --git a/third_party/slf4j-1.6.1/slf4j-log4j12-1.6.1.jar b/third_party/slf4j-1.6.1/slf4j-log4j12-1.6.1.jar
new file mode 100644
index 0000000000000000000000000000000000000000..873d11983e18b71aeafbc3d805495d6842339812
Binary files /dev/null and b/third_party/slf4j-1.6.1/slf4j-log4j12-1.6.1.jar differ