Skip to content

[SPARK-52334][CORE][K8S] update all files, jars, and pyFiles to reference the working directory after they are downloaded #51037

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 5 commits into from
Closed

[SPARK-52334][CORE][K8S] update all files, jars, and pyFiles to reference the working directory after they are downloaded #51037

Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
900b8e6
update all files, jars, archiveFiles, and pyFiles to reference the wo…
TongWei1105 May 28, 2025
416af57
fix
TongWei1105 May 28, 2025
8d6c3cd
add ut
TongWei1105 May 29, 2025
f2cd445
add ut for archives
TongWei1105 Jun 3, 2025
4195283
fix
TongWei1105 Jun 3, 2025
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
6 changes: 4 additions & 2 deletions core/src/main/scala/org/apache/spark/deploy/SparkSubmit.scala
View file
Original file line number Diff line number Diff line change
@@ -449,14 +449,16 @@ private[spark] class SparkSubmit extends Logging {
log" from ${MDC(LogKeys.SOURCE_PATH, source)}" +
log" to ${MDC(LogKeys.DESTINATION_PATH, dest)}")
Utils.deleteRecursively(dest)
if (isArchive) {
val resourceUri = if (isArchive) {
Utils.unpack(source, dest)
localResources
} else {
Files.copy(source.toPath, dest.toPath)
dest.toURI
}
// Keep the URIs of local files with the given fragments.
Utils.getUriBuilder(
localResources).fragment(resolvedUri.getFragment).build().toString
resourceUri).fragment(resolvedUri.getFragment).build().toString
} ++ avoidDownloads.map(_.toString)).mkString(",")
}

Binary file added core/src/test/resources/archive1.zip
View file
Binary file not shown.
37 changes: 37 additions & 0 deletions core/src/test/scala/org/apache/spark/deploy/SparkSubmitSuite.scala
View file
Original file line number Diff line number Diff line change
@@ -1809,6 +1809,43 @@ class SparkSubmitSuite
assert(classpath.contains("."))
}

test("SPARK-52334: Update all files, jars, and pyFiles to" +
"reference the working directory after they are downloaded") {
val testFile = "test_metrics_config.properties"
val testPyFile = "test_metrics_system.properties"
val testJar = "TestUDTF.jar"
val testArchives = "archive1.zip#test_archives"
val clArgs = Seq(
"--deploy-mode", "client",
"--proxy-user", "test.user",
"--master", "k8s://host:port",
"--executor-memory", "5g",
"--class", "org.SomeClass",
"--driver-memory", "4g",
"--conf", "spark.kubernetes.namespace=spark",
"--conf", "spark.kubernetes.driver.container.image=bar",
"--conf", "spark.kubernetes.submitInDriver=true",
"--files", s"src/test/resources/$testFile",
"--py-files", s"src/test/resources/$testPyFile",
"--jars", s"src/test/resources/$testJar",
"--archives", s"src/test/resources/$testArchives",
"/home/thejar.jar",
"arg1")
val appArgs = new SparkSubmitArguments(clArgs)
val _ = submit.prepareSubmitEnvironment(appArgs)
Copy link
Contributor

@mridulm mridulm Jun 2, 2025 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Can you update this test to handle archive as well ?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Can you update this test to handle archive as well ?

Done

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

log4j2.properties is not an archive file - and so ends up getting copied for destination (variant of existing cases).
I am trying to ensure that if (isArchive) { works as expected when the file actually results in unpacking the file

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

log4j2.properties is not an archive file - and so ends up getting copied for destination (variant of existing cases). I am trying to ensure that if (isArchive) { works as expected when the file actually results in unpacking the file

The isArchive logic does get triggered in this case — the test has been updated to cover that scenario accordingly.
Thank you for your suggestion.


appArgs.files should be (Utils.resolveURIs(s"$testFile,$testPyFile"))
appArgs.pyFiles should be (Utils.resolveURIs(testPyFile))
appArgs.jars should be (Utils.resolveURIs(testJar))
appArgs.archives should be (Utils.resolveURIs(s"src/test/resources/$testArchives"))

Files.isDirectory(Paths.get("test_archives")) should be(true)
Files.delete(Paths.get(testFile))
Files.delete(Paths.get(testPyFile))
Files.delete(Paths.get(testJar))
Files.delete(Paths.get("test_archives/archive1.text"))
}

// Requires Python dependencies for Spark Connect. Should be enabled by default.
ignore("Spark Connect application submission (Python)") {
val pyFile = File.createTempFile("remote_test", ".py")