diff --git a/core/src/main/scala/org/apache/spark/SparkContext.scala b/core/src/main/scala/org/apache/spark/SparkContext.scala
index ac7935b8c231ea51a47d08eb7764e6dc57150dc6..55602a90829d519159db1298f65fffe9396532c1 100644
--- a/core/src/main/scala/org/apache/spark/SparkContext.scala
+++ b/core/src/main/scala/org/apache/spark/SparkContext.scala
@@ -837,11 +837,12 @@ class SparkContext(config: SparkConf) extends Logging {
case "local" => "file:" + uri.getPath
case _ => path
}
- addedFiles(key) = System.currentTimeMillis
+ val timestamp = System.currentTimeMillis
+ addedFiles(key) = timestamp
// Fetch the file locally in case a job is executed using DAGScheduler.runLocally().
Utils.fetchFile(path, new File(SparkFiles.getRootDirectory()), conf, env.securityManager,
- hadoopConfiguration)
+ hadoopConfiguration, timestamp, useCache = false)
logInfo("Added file " + path + " at " + key + " with timestamp " + addedFiles(key))
postEnvironmentUpdate()
diff --git a/core/src/main/scala/org/apache/spark/executor/Executor.scala b/core/src/main/scala/org/apache/spark/executor/Executor.scala
index 616c7e6a4636825c48cf278c54434353b6981994..0b75b9b21fb821c50e7e8d2da9af29f473df3cf4 100644
--- a/core/src/main/scala/org/apache/spark/executor/Executor.scala
+++ b/core/src/main/scala/org/apache/spark/executor/Executor.scala
@@ -322,14 +322,16 @@ private[spark] class Executor(
// Fetch missing dependencies
for ((name, timestamp) <- newFiles if currentFiles.getOrElse(name, -1L) < timestamp) {
logInfo("Fetching " + name + " with timestamp " + timestamp)
- Utils.fetchFile(name, new File(SparkFiles.getRootDirectory), conf, env.securityManager,
- hadoopConf)
+ // Fetch file with useCache mode, close cache for local mode.
+ Utils.fetchFile(name, new File(SparkFiles.getRootDirectory), conf,
+ env.securityManager, hadoopConf, timestamp, useCache = !isLocal)
currentFiles(name) = timestamp
}
for ((name, timestamp) <- newJars if currentJars.getOrElse(name, -1L) < timestamp) {
logInfo("Fetching " + name + " with timestamp " + timestamp)
- Utils.fetchFile(name, new File(SparkFiles.getRootDirectory), conf, env.securityManager,
- hadoopConf)
+ // Fetch file with useCache mode, close cache for local mode.
+ Utils.fetchFile(name, new File(SparkFiles.getRootDirectory), conf,
+ env.securityManager, hadoopConf, timestamp, useCache = !isLocal)
currentJars(name) = timestamp
// Add it to our class loader
val localName = name.split("/").last
diff --git a/core/src/main/scala/org/apache/spark/util/Utils.scala b/core/src/main/scala/org/apache/spark/util/Utils.scala
index 0aeff6455b3fe266dd2ca590d0da115b7d918014..ccbddd985ae0abca727fd371555fd4561af84c64 100644
--- a/core/src/main/scala/org/apache/spark/util/Utils.scala
+++ b/core/src/main/scala/org/apache/spark/util/Utils.scala
@@ -347,15 +347,84 @@ private[spark] object Utils extends Logging {
}
/**
- * Download a file requested by the executor. Supports fetching the file in a variety of ways,
+ * Download a file to target directory. Supports fetching the file in a variety of ways,
+ * including HTTP, HDFS and files on a standard filesystem, based on the URL parameter.
+ *
+ * If `useCache` is true, first attempts to fetch the file to a local cache that's shared
+ * across executors running the same application. `useCache` is used mainly for
+ * the executors, and not in local mode.
+ *
+ * Throws SparkException if the target file already exists and has different contents than
+ * the requested file.
+ */
+ def fetchFile(
+ url: String,
+ targetDir: File,
+ conf: SparkConf,
+ securityMgr: SecurityManager,
+ hadoopConf: Configuration,
+ timestamp: Long,
+ useCache: Boolean) {
+ val fileName = url.split("/").last
+ val targetFile = new File(targetDir, fileName)
+ if (useCache) {
+ val cachedFileName = s"${url.hashCode}${timestamp}_cache"
+ val lockFileName = s"${url.hashCode}${timestamp}_lock"
+ val localDir = new File(getLocalDir(conf))
+ val lockFile = new File(localDir, lockFileName)
+ val raf = new RandomAccessFile(lockFile, "rw")
+ // Only one executor entry.
+ // The FileLock is only used to control synchronization for executors download file,
+ // it's always safe regardless of lock type (mandatory or advisory).
+ val lock = raf.getChannel().lock()
+ val cachedFile = new File(localDir, cachedFileName)
+ try {
+ if (!cachedFile.exists()) {
+ doFetchFile(url, localDir, cachedFileName, conf, securityMgr, hadoopConf)
+ }
+ } finally {
+ lock.release()
+ }
+ if (targetFile.exists && !Files.equal(cachedFile, targetFile)) {
+ if (conf.getBoolean("spark.files.overwrite", false)) {
+ targetFile.delete()
+ logInfo((s"File $targetFile exists and does not match contents of $url, " +
+ s"replacing it with $url"))
+ } else {
+ throw new SparkException(s"File $targetFile exists and does not match contents of $url")
+ }
+ }
+ Files.copy(cachedFile, targetFile)
+ } else {
+ doFetchFile(url, targetDir, fileName, conf, securityMgr, hadoopConf)
+ }
+
+ // Decompress the file if it's a .tar or .tar.gz
+ if (fileName.endsWith(".tar.gz") || fileName.endsWith(".tgz")) {
+ logInfo("Untarring " + fileName)
+ Utils.execute(Seq("tar", "-xzf", fileName), targetDir)
+ } else if (fileName.endsWith(".tar")) {
+ logInfo("Untarring " + fileName)
+ Utils.execute(Seq("tar", "-xf", fileName), targetDir)
+ }
+ // Make the file executable - That's necessary for scripts
+ FileUtil.chmod(targetFile.getAbsolutePath, "a+x")
+ }
+
+ /**
+ * Download a file to target directory. Supports fetching the file in a variety of ways,
* including HTTP, HDFS and files on a standard filesystem, based on the URL parameter.
*
* Throws SparkException if the target file already exists and has different contents than
* the requested file.
*/
- def fetchFile(url: String, targetDir: File, conf: SparkConf, securityMgr: SecurityManager,
- hadoopConf: Configuration) {
- val filename = url.split("/").last
+ private def doFetchFile(
+ url: String,
+ targetDir: File,
+ filename: String,
+ conf: SparkConf,
+ securityMgr: SecurityManager,
+ hadoopConf: Configuration) {
val tempDir = getLocalDir(conf)
val tempFile = File.createTempFile("fetchFileTemp", null, new File(tempDir))
val targetFile = new File(targetDir, filename)
@@ -443,16 +512,6 @@ private[spark] object Utils extends Logging {
}
Files.move(tempFile, targetFile)
}
- // Decompress the file if it's a .tar or .tar.gz
- if (filename.endsWith(".tar.gz") || filename.endsWith(".tgz")) {
- logInfo("Untarring " + filename)
- Utils.execute(Seq("tar", "-xzf", filename), targetDir)
- } else if (filename.endsWith(".tar")) {
- logInfo("Untarring " + filename)
- Utils.execute(Seq("tar", "-xf", filename), targetDir)
- }
- // Make the file executable - That's necessary for scripts
- FileUtil.chmod(targetFile.getAbsolutePath, "a+x")
}
/**