Delete Spark's temporary directories when the JVM exits.

core/src/main/scala/spark/ShuffleManager.scala

@@ -47,6 +47,13 @@ class ShuffleManager extends Logging {
     shuffleDir = new File(localDir, "shuffle")
     shuffleDir.mkdirs()
     logInfo("Shuffle dir: " + shuffleDir)
+
+    // Add a shutdown hook to delete the local dir
+    Runtime.getRuntime.addShutdownHook(new Thread("delete Spark local dir") {
+      override def run() {
+        Utils.deleteRecursively(localDir)
+      }
+    })
     
     val extServerPort = System.getProperty(
       "spark.localFileShuffle.external.server.port", "-1").toInt

core/src/main/scala/spark/Utils.scala

@@ -116,12 +116,12 @@ object Utils {
   }
 
   /**
-   * Get the local host's IP address in dotted-quad format (e.g. 1.2.3.4)
+   * Get the local host's IP address in dotted-quad format (e.g. 1.2.3.4).
    */
   def localIpAddress(): String = InetAddress.getLocalHost.getHostAddress
   
   /**
-   * Returns a standard ThreadFactory except all threads are daemons
+   * Returns a standard ThreadFactory except all threads are daemons.
    */
   private def newDaemonThreadFactory: ThreadFactory = {
     new ThreadFactory {
@@ -134,7 +134,7 @@ object Utils {
   }
 
   /**
-   * Wrapper over newCachedThreadPool
+   * Wrapper over newCachedThreadPool.
    */
   def newDaemonCachedThreadPool(): ThreadPoolExecutor = {
     var threadPool =
@@ -146,7 +146,7 @@ object Utils {
   }
 
   /**
-   * Wrapper over newFixedThreadPool
+   * Wrapper over newFixedThreadPool.
    */
   def newDaemonFixedThreadPool(nThreads: Int): ThreadPoolExecutor = {
     var threadPool =
@@ -158,9 +158,23 @@ object Utils {
   }
 
   /**
-   * Get the local machine's hostname
+   * Get the local machine's hostname.
    */
   def localHostName(): String = {
     return InetAddress.getLocalHost().getHostName
   }
+
+  /**
+   * Delete a file or directory and its contents recursively.
+   */
+  def deleteRecursively(file: File) {
+    if (file.isDirectory) {
+      for (child <- file.listFiles()) {
+        deleteRecursively(child)
+      }
+    }
+    if (!file.delete()) {
+      throw new IOException("Failed to delete: " + file)
+    }
+  }
 }