Skip to content

[SPARK-8657] [YARN] Simplify method addResource() and get the correct fs from destPath. #7053

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
wants to merge 2 commits into from
Closed

[SPARK-8657] [YARN] Simplify method addResource() and get the correct fs from destPath. #7053

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
75b457e
[SPARK-8657] [YARN] Simplify method addResource() and get the correct…
litao-buptsse Jun 27, 2015
ecdd811
Merge remote-tracking branch 'origin/master' into SPARK-8657-for-master
litao-buptsse Jun 30, 2015
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion yarn/src/main/scala/org/apache/spark/deploy/yarn/Client.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -322,7 +322,7 @@ private[spark] class Client(
destName.orElse(Option(localURI.getFragment())).getOrElse(localPath.getName())
val destPath = copyFileToRemote(dst, localPath, replication)
distCacheMgr.addResource(
fs, hadoopConf, destPath, localResources, resType, linkname, statCache,
hadoopConf, destPath, localResources, resType, linkname, statCache,
appMasterOnly = appMasterOnly)
(false, linkname)
} else {
Expand Down
3 changes: 1 addition & 2 deletions yarn/src/main/scala/org/apache/spark/deploy/yarn/ClientDistributedCacheManager.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -46,7 +46,6 @@ private[spark] class ClientDistributedCacheManager() extends Logging {
* Adds the LocalResource to the localResources HashMap passed in and saves
* the stats of the resources to they can be sent to the executors and verified.
*
* @param fs FileSystem
* @param conf Configuration
* @param destPath path to the resource
* @param localResources localResource hashMap to insert the resource into
Expand All @@ -56,14 +55,14 @@ private[spark] class ClientDistributedCacheManager() extends Logging {
* @param appMasterOnly Whether to only add the resource to the app master
*/
def addResource(
fs: FileSystem,
conf: Configuration,
destPath: Path,
localResources: HashMap[String, LocalResource],
resourceType: LocalResourceType,
link: String,
statCache: Map[URI, FileStatus],
appMasterOnly: Boolean = false): Unit = {
val fs = FileSystem.get(destPath.toUri(), conf)
val destStatus = fs.getFileStatus(destPath)
val amJarRsrc = Records.newRecord(classOf[LocalResource])
amJarRsrc.setType(resourceType)
Expand Down
10 changes: 5 additions & 5 deletions yarn/src/test/scala/org/apache/spark/deploy/yarn/ClientDistributedCacheManagerSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -78,7 +78,7 @@ class ClientDistributedCacheManagerSuite extends SparkFunSuite with MockitoSugar
val statCache: Map[URI, FileStatus] = HashMap[URI, FileStatus]()
when(fs.getFileStatus(destPath)).thenReturn(new FileStatus())

distMgr.addResource(fs, conf, destPath, localResources, LocalResourceType.FILE, "link",
distMgr.addResource(conf, destPath, localResources, LocalResourceType.FILE, "link",
statCache, false)
val resource = localResources("link")
assert(resource.getVisibility() === LocalResourceVisibility.PRIVATE)
Expand All @@ -105,7 +105,7 @@ class ClientDistributedCacheManagerSuite extends SparkFunSuite with MockitoSugar
null, new Path("/tmp/testing2"))
val destPath2 = new Path("file:///foo.invalid.com:8080/tmp/testing2")
when(fs.getFileStatus(destPath2)).thenReturn(realFileStatus)
distMgr.addResource(fs, conf, destPath2, localResources, LocalResourceType.FILE, "link2",
distMgr.addResource(conf, destPath2, localResources, LocalResourceType.FILE, "link2",
statCache, false)
val resource2 = localResources("link2")
assert(resource2.getVisibility() === LocalResourceVisibility.PRIVATE)
Expand Down Expand Up @@ -141,7 +141,7 @@ class ClientDistributedCacheManagerSuite extends SparkFunSuite with MockitoSugar
when(fs.getFileStatus(destPath)).thenReturn(new FileStatus())

intercept[Exception] {
distMgr.addResource(fs, conf, destPath, localResources, LocalResourceType.FILE, null,
distMgr.addResource(conf, destPath, localResources, LocalResourceType.FILE, null,
statCache, false)
}
assert(localResources.get("link") === None)
Expand All @@ -159,7 +159,7 @@ class ClientDistributedCacheManagerSuite extends SparkFunSuite with MockitoSugar
null, new Path("/tmp/testing"))
when(fs.getFileStatus(destPath)).thenReturn(realFileStatus)

distMgr.addResource(fs, conf, destPath, localResources, LocalResourceType.ARCHIVE, "link",
distMgr.addResource(conf, destPath, localResources, LocalResourceType.ARCHIVE, "link",
statCache, true)
val resource = localResources("link")
assert(resource.getVisibility() === LocalResourceVisibility.PRIVATE)
Expand Down Expand Up @@ -193,7 +193,7 @@ class ClientDistributedCacheManagerSuite extends SparkFunSuite with MockitoSugar
null, new Path("/tmp/testing"))
when(fs.getFileStatus(destPath)).thenReturn(realFileStatus)

distMgr.addResource(fs, conf, destPath, localResources, LocalResourceType.ARCHIVE, "link",
distMgr.addResource(conf, destPath, localResources, LocalResourceType.ARCHIVE, "link",
statCache, false)
val resource = localResources("link")
assert(resource.getVisibility() === LocalResourceVisibility.PRIVATE)
Expand Down