[SPARK-2713] Executors of same application in same host should only download files & jars once #1616

li-zhihui · 2014-07-28T05:32:31Z

If Spark lunched multiple executors in one host for one application, every executor would download it dependent files and jars (if not using local: url) independently. It maybe result in huge latency. In my case, it result in 20 seconds latency to download dependent jars(size about 17M) when I lunched 32 executors in every host(total 4 hosts).

This patch will cache downloaded files and jars for executors to reduce network throughput and download latency. In my case, the latency was reduced from 20 seconds to less than 1 second.

AmplabJenkins · 2014-07-28T05:37:18Z

Can one of the admins verify this patch?

harishreedharan · 2014-07-28T05:52:00Z

core/src/main/scala/org/apache/spark/util/Utils.scala

+      fetchFile(url, localDir, conf, securityMgr)
+      Files.move(new File(localDir, fileName), cachedFile)
+    }
+    lock.release()


If the move throws some exception, the lock may never be released. You should wrap the release call in a finally block.

thanks @harishreedharan done.

JoshRosen · 2014-07-29T23:18:09Z

Jenkins, this is ok to test.

SparkQA · 2014-07-29T23:23:49Z

QA tests have started for PR 1616. This patch merges cleanly.
View progress: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/17391/consoleFull

SparkQA · 2014-07-30T00:12:11Z

QA results for PR 1616:
- This patch PASSES unit tests.
- This patch merges cleanly
- This patch adds no public classes

For more information see test ouptut:
https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/17391/consoleFull

li-zhihui · 2014-08-04T02:07:14Z

@JoshRosen more comments?

JoshRosen · 2014-08-04T05:30:56Z

This uses FileLock as its locking mechanism. According to those docs (emphasis mine),

Whether or not a lock actually prevents another program from accessing the content of the locked region is system-dependent and therefore unspecified. The native file-locking facilities of some systems are merely advisory, meaning that programs must cooperatively observe a known locking protocol in order to guarantee data integrity. On other systems native file locks are mandatory, meaning that if one program locks a region of a file then other programs are actually prevented from accessing that region in a way that would violate the lock. On yet other systems, whether native file locks are advisory or mandatory is configurable on a per-file basis. To ensure consistent and correct behavior across platforms, it is strongly recommended that the locks provided by this API be used as if they were advisory locks.

Can you comment on whether this approach is safe if we're using advisory locks, and maybe add that comment to the source code?

SparkQA · 2014-08-04T09:44:23Z

QA tests have started for PR 1616. This patch merges cleanly.
View progress: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/17852/consoleFull

SparkQA · 2014-08-04T10:41:59Z

QA results for PR 1616:
- This patch PASSES unit tests.
- This patch merges cleanly
- This patch adds no public classes

For more information see test ouptut:
https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/17852/consoleFull

li-zhihui · 2014-08-04T12:09:54Z

@JoshRosen added comment.

JoshRosen · 2014-08-04T16:19:00Z

Thanks for commenting. I now realize that my concern about advisory locking was a little misguided, since only cooperating Spark processes will be coordinating through the lock file.

JoshRosen · 2014-08-04T16:48:21Z

This seems like an alright fix and I'd like to get it into a release, but I'm concerned that this doesn't correctly handle every possible feature of fetchFile.

For example, there's some code in fetchFile to automatically decompress .tar.gz files. I don't remember why this code was added (or whether it's actually correct, since it seems to assume that files are downloaded into the current working directory), but I'm not sure that fetchCachedFile will properly handle that case; it seems like it would only copy the .tar.gz file without decompressing it in the executor's directory.

We could try to special-case fix this by moving the decompression logic into fetchCachedFile, but I'm worried that it will make fetchFile even harder to understand. I think that fetchFile might be due for a refactoring.

Also, do you think we should just replace fetchFile with fetchCachedFile and keep the uncached version private?

SparkQA · 2014-08-05T02:39:24Z

QA tests have started for PR 1616. This patch merges cleanly.
View progress: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/17915/consoleFull

SparkQA · 2014-08-05T03:47:10Z

QA results for PR 1616:
- This patch PASSES unit tests.
- This patch merges cleanly
- This patch adds no public classes

For more information see test ouptut:
https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/17915/consoleFull

li-zhihui · 2014-08-05T04:08:46Z

Thaks @JoshRosen sorry I missed the important operation (and I missed FileUtil.chmod(targetFile.getAbsolutePath, "a+x") too).

I add a new commit.

li-zhihui · 2014-08-21T03:01:58Z

@JoshRosen any more comments?

JoshRosen · 2014-08-21T07:24:51Z

Thanks a bunch for updating this; this seems like an important fix and I'd like to try to get it included soon in a release. I'll try my best to review this tomorrow and merge it if it looks good.

li-zhihui · 2014-08-27T02:29:25Z

@JoshRosen do you have time to review it?

pwendell · 2014-08-27T03:27:31Z

@JoshRosen if you do merge this please only into master and not 1.1... we are only fixing major regressions in 1.1 right now.

JoshRosen · 2014-09-03T22:03:11Z

core/src/main/scala/org/apache/spark/util/Utils.scala

@@ -317,13 +317,58 @@ private[spark] object Utils extends Logging {
  }

  /**
+   * Copy cached file to targetDir, if not exists, download it from url firstly.


Minor nitpick on naming, but I think it's confusing to have a method named fetchCachedFile with an option that has to be explicitly set in order to use the cache. I'd prefer to name this fetchFile, and rename the other method to something like doFetchFile or _fetchFile.

When fixing the merge conflict, do you mind moving the comment from the old fetchFile to here? I think the most comprehensive documentation should be on the public function, not the private one. I'd say something like

/** * Download a file requested by the executor . 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 from a local cache that's shared across * executors running the same application. * * Throws SparkException if the target file already exists and has different contents than * the requested file. */

JoshRosen · 2014-09-03T22:12:28Z

Hey, sorry to drop the ball on this review. Things got really busy during the 1.1.0 QA process, but I'm slowly getting back to my reviews now.

Do you think there's any potential for conflicts between multiple applications that attempt to add files with the same name but different contents? Different applications will share the same local directory. Maybe the timestamp takes care of this, assuming that it's unlikely for us to have timestamp collisions.

/cc @andrewor14, do you have any thoughts on this?

JoshRosen · 2014-09-03T22:17:54Z

Actually, I don't think the timestamp will help us here:

If app A and B simultaneously add files named foo.txt and simultaneously attempt to download this file on the same worker (from different executors), then both will see that the cached file doesn't exist and both will attempt to download the file by calling the old fetchFile with the same name and target directory. This creates a race condition, since they're both attempting to write different contents to the same targetFile.

andrewor14 · 2014-09-03T22:48:07Z

core/src/main/scala/org/apache/spark/util/Utils.scala

+   * If useCache == false, download file to targetDir directly.
+   */
+  def fetchCachedFile(url: String, targetDir: File, conf: SparkConf, securityMgr: SecurityManager,
+    timestamp: Long, useCache: Boolean) {


style should be

def fetchCachedFile( url: String, targetDir: File, ... useCache: Boolean) { ... }

andrewor14 · 2014-09-03T22:53:57Z

Yes, it does seem like a problem if multiple simultaneous applications share the same files. Do we handle that even before this patch? I haven't dug deep into this, but should we have some kind application-specific directory for fetching files?

JoshRosen · 2014-09-03T23:02:47Z

@andrewor14 I don't think that it was a problem before, but the reason is perhaps a little subtle:

The old fetchFile has a workflow where it first downloads the file to a temporary file and then moves that temporary file to its final destination. Although the parent directory of the temporary files (spark.local.dir) is shared by all executors, the actual temporary file is created through File.createTempFile, so it should have a unique name. After downloading the file, fetchFile moves it to targetDir and renames it. When fetching a file on an executor, targetDir is SparkFiles.getRootDirectory, which is a per-application temporary directory, so there's no potential for cross-application conflicts.

This PR uses that same code path to perform the actual download. The potential conflict occurs because targetDir is localDir when downloading a file that's not present in the cache.

li-zhihui · 2014-09-04T05:19:40Z

@JoshRosen @andrewor14

I test the patch in yarn mode, and the localDir is a per-application temporary directory in this mode. Now I know it is a problem in standalone(and mesos) mode.
targetDir(from SparkFiles.getRootDirectory) is per-executor temporary directory(in my case ,it is /home/frank/hdfs/yarn/nm-local-dir/usercache/frank/appcache/application_1409795343243_0002/container_1409795343243_0002_01_000126/./), I think we can use targetDir + "../" as per-application directory to save the cache file.(abandon the solution)

BTW: The timestamp follow this code's logic: https://github.com/apache/spark/blob/master/core/src/main/scala/org/apache/spark/executor/Executor.scala#L323 Although I don't understand why the timestamp could be changed in an application's life time.

… & jars once

andrewor14 · 2014-10-01T02:53:48Z

core/src/main/scala/org/apache/spark/util/Utils.scala

   * 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 the executors, not in local mode.


"and" not in local mode

Done, thanks.

SparkQA · 2014-10-08T05:24:32Z

QA tests have started for PR 1616 at commit 935fed6.

This patch merges cleanly.

SparkQA · 2014-10-08T06:27:52Z

QA tests have finished for PR 1616 at commit 935fed6.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

AmplabJenkins · 2014-10-08T06:27:55Z

Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/21447/Test PASSed.

li-zhihui · 2014-10-22T01:48:00Z

@andrewor14 more comments?

andrewor14 · 2014-10-22T22:59:08Z

core/src/main/scala/org/apache/spark/executor/Executor.scala

-        Utils.fetchFile(name, new File(SparkFiles.getRootDirectory), conf, env.securityManager,
-          hadoopConf)
+        Utils.fetchFile(name, new File(SparkFiles.getRootDirectory), conf,
+          env.securityManager, hadoopConf, timestamp, useCache = true)


I just noticed, if this isn't meant to be used in local mode, shouldn't this be useCache = !isLocal?

If so, it would be good if you could add a small comment to explain here that the cache is not needed for local mode because there is no fetching involved.

thanks @andrewor14 , done

andrewor14 · 2014-10-22T23:05:18Z

Hey yeah @li-zhihui sorry this slipped on our end. This LGTM except for the one comment I made just now. I think the intention is that executors running in local mode shouldn't have to use the cache, as expressed in your javadoc for fetchFile. The code does otherwise, however.

SparkQA · 2014-10-24T01:05:01Z

QA tests have started for PR 1616 at commit 36940df.

This patch merges cleanly.

SparkQA · 2014-10-24T03:05:01Z

Tests timed out for PR 1616 at commit 36940df after a configured wait of 120m.

AmplabJenkins · 2014-10-24T03:05:04Z

Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/22101/
Test FAILed.

andrewor14 · 2014-10-24T03:15:28Z

retest this please

SparkQA · 2014-10-24T03:19:47Z

QA tests have started for PR 1616 at commit 36940df.

This patch merges cleanly.

SparkQA · 2014-10-24T04:23:05Z

QA tests have finished for PR 1616 at commit 36940df.

This patch fails PySpark unit tests.
This patch merges cleanly.
This patch adds no public classes.

AmplabJenkins · 2014-10-24T04:23:09Z

Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/22108/
Test FAILed.

li-zhihui · 2014-10-24T05:25:57Z

@andrewor14 I guess the failure is non-interrelated with the patch. But I don't know why failed again, can you give me some advice?

andrewor14 · 2014-10-24T17:56:29Z

retest this please

andrewor14 · 2014-10-24T17:56:56Z

Yeah pyspark tests are kinda flaky. There's no way this patch could have caused it.

SparkQA · 2014-10-24T17:59:41Z

Test build #22147 has started for PR 1616 at commit 36940df.

This patch merges cleanly.

SparkQA · 2014-10-24T19:08:51Z

Test build #22147 has finished for PR 1616 at commit 36940df.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

AmplabJenkins · 2014-10-24T19:08:54Z

Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/22147/
Test PASSed.

andrewor14 · 2014-10-24T20:00:37Z

Ok cool I'm merging this. Thanks @li-zhihui

harishreedharan reviewed Jul 28, 2014
View reviewed changes

JoshRosen reviewed Sep 3, 2014
View reviewed changes

andrewor14 reviewed Sep 3, 2014
View reviewed changes

li-zhihui added 2 commits September 4, 2014 14:39

Executors of same application in same host should only download files…

6b997bf

… & jars once

Release lock before copy files

7fb7c0b

andrewor14 reviewed Oct 1, 2014
View reviewed changes

Clean code.

935fed6

andrewor14 reviewed Oct 22, 2014
View reviewed changes

Close cache for local mode

36940df

asfgit closed this in 7aacb7b Oct 24, 2014

JoshRosen mentioned this pull request Feb 3, 2015

SPARK-4687. Add a recursive option to the addFile API #3670

Closed

[SPARK-2713] Executors of same application in same host should only download files & jars once #1616

[SPARK-2713] Executors of same application in same host should only download files & jars once #1616

Uh oh!

Conversation

li-zhihui commented Jul 28, 2014

Uh oh!

AmplabJenkins commented Jul 28, 2014

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

JoshRosen commented Jul 29, 2014

Uh oh!

SparkQA commented Jul 29, 2014

Uh oh!

SparkQA commented Jul 30, 2014

Uh oh!

li-zhihui commented Aug 4, 2014

Uh oh!

JoshRosen commented Aug 4, 2014

Uh oh!

SparkQA commented Aug 4, 2014

Uh oh!

SparkQA commented Aug 4, 2014

Uh oh!

li-zhihui commented Aug 4, 2014

Uh oh!

JoshRosen commented Aug 4, 2014

Uh oh!

JoshRosen commented Aug 4, 2014

Uh oh!

SparkQA commented Aug 5, 2014

Uh oh!

SparkQA commented Aug 5, 2014

Uh oh!

li-zhihui commented Aug 5, 2014

Uh oh!

li-zhihui commented Aug 21, 2014

Uh oh!

JoshRosen commented Aug 21, 2014

Uh oh!

li-zhihui commented Aug 27, 2014

Uh oh!

pwendell commented Aug 27, 2014

Uh oh!

Choose a reason for hiding this comment

Uh oh!

JoshRosen commented Sep 3, 2014

Uh oh!

JoshRosen commented Sep 3, 2014

Uh oh!

Choose a reason for hiding this comment

Uh oh!

andrewor14 commented Sep 3, 2014

Uh oh!

JoshRosen commented Sep 3, 2014

Uh oh!

li-zhihui commented Sep 4, 2014

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

SparkQA commented Oct 8, 2014

Uh oh!

SparkQA commented Oct 8, 2014

Uh oh!

AmplabJenkins commented Oct 8, 2014

Uh oh!

li-zhihui commented Oct 22, 2014

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!