dongjoon-hyun commented on code in PR #53845:
URL: https://github.com/apache/spark/pull/53845#discussion_r4000555453
##########
core/src/main/scala/org/apache/spark/deploy/SparkSubmit.scala:
##########
@@ -469,14 +469,20 @@ private[spark] class SparkSubmit extends Logging {
def avoidJarDownload(scheme: String): Boolean =
avoidJarDownloadSchemes.contains("*") ||
avoidJarDownloadSchemes.contains(scheme)
+ val avoidArchiveDownloadSchemes =
sparkConf.get(KUBERNETES_ARCHIVES_AVOID_DOWNLOAD_SCHEMES)
+
+ def avoidArchiveDownload(scheme: String): Boolean =
+ avoidArchiveDownloadSchemes.contains("*") ||
avoidArchiveDownloadSchemes.contains(scheme)
Review Comment:
Since `*` also matches `file` and `local`, the archives already inside the
driver image are not unpacked into the working directory either, although there
is nothing to download for them (`DependencyUtils.downloadFile` returns
`file`/`local` paths as they are).
- `local:` archives are not unpacked anywhere on the driver because
`SparkContext.addFile` skips the `local` scheme.
- `file:` archives are still served to the executors via the driver file
server, so `*` doesn't help the network saturation for them.
Is this intended? If not, we may want to exclude `file` and `local` here.
##########
core/src/main/scala/org/apache/spark/deploy/SparkSubmit.scala:
##########
@@ -469,14 +469,20 @@ private[spark] class SparkSubmit extends Logging {
def avoidJarDownload(scheme: String): Boolean =
avoidJarDownloadSchemes.contains("*") ||
avoidJarDownloadSchemes.contains(scheme)
+ val avoidArchiveDownloadSchemes =
sparkConf.get(KUBERNETES_ARCHIVES_AVOID_DOWNLOAD_SCHEMES)
+
+ def avoidArchiveDownload(scheme: String): Boolean =
Review Comment:
This is identical to `avoidJarDownload` except the config. Can we have a
single helper, e.g. `def matchesSchemes(schemes: Seq[String])(scheme: String):
Boolean`, and use it for both? Otherwise, we need another copy whenever `files`
or `pyFiles` need the same option.
##########
core/src/test/scala/org/apache/spark/deploy/SparkSubmitSuite.scala:
##########
@@ -605,6 +605,109 @@ class SparkSubmitSuite
}
}
+ test("SPARK-55077: Avoid archives download if scheme matches " +
+ "spark.kubernetes.archives.avoidDownloadSchemes " +
+ "in k8s client mode & driver runs inside a POD") {
+ val hadoopConf = new Configuration()
+ updateConfWithFakeS3Fs(hadoopConf)
+ withTempDir { tmpDir =>
+ val notToDownloadArchive = File.createTempFile("NotToDownloadArchive",
".zip", tmpDir)
+ val remoteArchiveFile = s"s3a://${notToDownloadArchive.getAbsolutePath}"
+ val tmpJar = File.createTempFile("TestUDTF", ".jar", tmpDir)
Review Comment:
Do we need `tmpJar` here? If the intention is to verify that the other
resources are still downloaded, `--files` and `--py-files` already cover it. In
addition, the three new test cases are almost identical except the config value
and the expectations, so a shared helper would reduce the duplication.
##########
core/src/main/scala/org/apache/spark/deploy/SparkSubmit.scala:
##########
@@ -469,14 +469,20 @@ private[spark] class SparkSubmit extends Logging {
def avoidJarDownload(scheme: String): Boolean =
avoidJarDownloadSchemes.contains("*") ||
avoidJarDownloadSchemes.contains(scheme)
+ val avoidArchiveDownloadSchemes =
sparkConf.get(KUBERNETES_ARCHIVES_AVOID_DOWNLOAD_SCHEMES)
+
+ def avoidArchiveDownload(scheme: String): Boolean =
+ avoidArchiveDownloadSchemes.contains("*") ||
avoidArchiveDownloadSchemes.contains(scheme)
+
val filesLocalFiles = Option(args.files).map {
downloadResourcesToCurrentDirectory(_)
}.orNull
val updatedJars = Option(args.jars).map {
downloadResourcesToCurrentDirectory(_, avoidDownload =
avoidJarDownload)
}.orNull
val archiveLocalFiles = Option(args.archives).map {
- downloadResourcesToCurrentDirectory(_, true)
+ downloadResourcesToCurrentDirectory(_, isArchive = true,
+ avoidDownload = avoidArchiveDownload)
Review Comment:
With this option, the matched archives are no longer unpacked into the
driver's current working directory. So, the SPARK-33748 behavior (archives are
available in the driver's working directory in cluster mode) is silently lost.
This breaks the common PySpark packaging pattern on K8s. For example,
```
--archives s3a://bucket/pyspark_env.tar.gz#environment
--conf spark.pyspark.python=./environment/bin/python
--conf spark.kubernetes.archives.avoidDownloadSchemes=s3a
```
`PythonRunner` launches `./environment/bin/python` before `SparkContext` is
created, so the driver fails with `Cannot run program
"./environment/bin/python"`. The archive is unpacked only later under
`SparkFiles.getRootDirectory()` by `SparkContext.addFile`.
Since the driver downloads the archive in `SparkContext.addFile` anyway, the
executor fan-in actually comes from rewriting `spark.archives` to the
driver-local `file:` path (`localResources`). Could we keep unpacking into the
working directory and keep the original remote URI in `spark.archives` only for
the matched schemes? If not, we need to document this limitation at least.
##########
core/src/test/scala/org/apache/spark/deploy/SparkSubmitSuite.scala:
##########
@@ -605,6 +605,109 @@ class SparkSubmitSuite
}
}
+ test("SPARK-55077: Avoid archives download if scheme matches " +
+ "spark.kubernetes.archives.avoidDownloadSchemes " +
+ "in k8s client mode & driver runs inside a POD") {
+ val hadoopConf = new Configuration()
+ updateConfWithFakeS3Fs(hadoopConf)
+ withTempDir { tmpDir =>
+ val notToDownloadArchive = File.createTempFile("NotToDownloadArchive",
".zip", tmpDir)
+ val remoteArchiveFile = s"s3a://${notToDownloadArchive.getAbsolutePath}"
+ val tmpJar = File.createTempFile("TestUDTF", ".jar", tmpDir)
+
+ val clArgs = Seq(
+ "--deploy-mode", "client",
+ "--proxy-user", "test.user",
+ "--master", "k8s://host:port",
+ "--class", "org.SomeClass",
+ "--conf", "spark.kubernetes.submitInDriver=true",
+ "--conf", "spark.kubernetes.archives.avoidDownloadSchemes=s3a",
+ "--conf",
"spark.hadoop.fs.s3a.impl=org.apache.spark.deploy.TestFileSystem",
+ "--conf", "spark.hadoop.fs.s3a.impl.disable.cache=true",
+ "--files", "src/test/resources/test_metrics_config.properties",
+ "--py-files", "src/test/resources/test_metrics_system.properties",
+ "--archives",
s"src/test/resources/log4j2.properties,$remoteArchiveFile",
+ "--jars", tmpJar.getAbsolutePath,
+ "/home/jarToIgnore.jar",
+ "arg1")
+ val appArgs = new SparkSubmitArguments(clArgs)
+ val (_, _, conf, _) = submit.prepareSubmitEnvironment(appArgs,
Some(hadoopConf))
+ conf.get("spark.master") should be("k8s://https://host:port";)
+ conf.get("spark.archives").contains(remoteArchiveFile) shouldBe true
+ conf.get("spark.archives").contains("log4j2.properties") shouldBe true
+
+ Files.exists(Paths.get("test_metrics_config.properties")) should be(true)
+ Files.exists(Paths.get("test_metrics_system.properties")) should be(true)
+ Files.exists(Paths.get("log4j2.properties")) should be(true)
+ Files.exists(Paths.get(tmpJar.getName)) should be(true)
+ Files.exists(Paths.get(notToDownloadArchive.getName)) should be(false)
+ Files.delete(Paths.get("test_metrics_config.properties"))
+ Files.delete(Paths.get("test_metrics_system.properties"))
+ Files.delete(Paths.get("log4j2.properties"))
+ Files.delete(Paths.get(tmpJar.getName))
+ }
+ }
+
+ test("SPARK-55077: Avoid archives download for wildcard scheme " +
+ "spark.kubernetes.archives.avoidDownloadSchemes " +
+ "in k8s client mode & driver runs inside a POD") {
+ val hadoopConf = new Configuration()
+ updateConfWithFakeS3Fs(hadoopConf)
+ withTempDir { tmpDir =>
+ val notToDownloadArchive =
File.createTempFile("NotToDownloadArchiveWildcard", ".zip", tmpDir)
+ val remoteArchiveFile = s"s3a://${notToDownloadArchive.getAbsolutePath}"
+
+ val clArgs = Seq(
+ "--deploy-mode", "client",
+ "--proxy-user", "test.user",
+ "--master", "k8s://host:port",
+ "--class", "org.SomeClass",
+ "--conf", "spark.kubernetes.submitInDriver=true",
+ "--conf", "spark.kubernetes.archives.avoidDownloadSchemes=*",
+ "--conf",
"spark.hadoop.fs.s3a.impl=org.apache.spark.deploy.TestFileSystem",
+ "--conf", "spark.hadoop.fs.s3a.impl.disable.cache=true",
+ "--archives", remoteArchiveFile,
+ "/home/jarToIgnore.jar",
+ "arg1")
+ val appArgs = new SparkSubmitArguments(clArgs)
+ val (_, _, conf, _) = submit.prepareSubmitEnvironment(appArgs,
Some(hadoopConf))
+ conf.get("spark.master") should be("k8s://https://host:port";)
+ conf.get("spark.archives").contains(remoteArchiveFile) shouldBe true
+ Files.exists(Paths.get(notToDownloadArchive.getName)) should be(false)
+ }
+ }
+
+ test("SPARK-55077: Download archives if scheme does not match " +
+ "spark.kubernetes.archives.avoidDownloadSchemes " +
+ "in k8s client mode & driver runs inside a POD") {
+ val hadoopConf = new Configuration()
+ updateConfWithFakeS3Fs(hadoopConf)
+ withTempDir { tmpDir =>
+ val toDownloadArchive = File.createTempFile("ToDownloadArchive",
".properties", tmpDir)
+ Files.write(toDownloadArchive.toPath, Array[Byte](1, 2, 3))
+ val remoteArchiveFile = s"s3a://${toDownloadArchive.getAbsolutePath}"
+
+ val clArgs = Seq(
+ "--deploy-mode", "client",
+ "--proxy-user", "test.user",
+ "--master", "k8s://host:port",
+ "--class", "org.SomeClass",
+ "--conf", "spark.kubernetes.submitInDriver=true",
+ "--conf", "spark.kubernetes.archives.avoidDownloadSchemes=hdfs",
+ "--conf",
"spark.hadoop.fs.s3a.impl=org.apache.spark.deploy.TestFileSystem",
+ "--conf", "spark.hadoop.fs.s3a.impl.disable.cache=true",
+ "--archives", remoteArchiveFile,
+ "/home/jarToIgnore.jar",
+ "arg1")
+ val appArgs = new SparkSubmitArguments(clArgs)
+ val (_, _, conf, _) = submit.prepareSubmitEnvironment(appArgs,
Some(hadoopConf))
+ conf.get("spark.master") should be("k8s://https://host:port";)
+ conf.get("spark.archives").contains(remoteArchiveFile) shouldBe false
Review Comment:
This only checks that the remote URI is absent, so it passes even if the
archive is dropped from `spark.archives`. Can we assert that `spark.archives`
contains the downloaded local file instead?
##########
core/src/test/scala/org/apache/spark/deploy/SparkSubmitSuite.scala:
##########
@@ -605,6 +605,109 @@ class SparkSubmitSuite
}
}
+ test("SPARK-55077: Avoid archives download if scheme matches " +
+ "spark.kubernetes.archives.avoidDownloadSchemes " +
+ "in k8s client mode & driver runs inside a POD") {
+ val hadoopConf = new Configuration()
+ updateConfWithFakeS3Fs(hadoopConf)
+ withTempDir { tmpDir =>
+ val notToDownloadArchive = File.createTempFile("NotToDownloadArchive",
".zip", tmpDir)
+ val remoteArchiveFile = s"s3a://${notToDownloadArchive.getAbsolutePath}"
+ val tmpJar = File.createTempFile("TestUDTF", ".jar", tmpDir)
+
+ val clArgs = Seq(
+ "--deploy-mode", "client",
+ "--proxy-user", "test.user",
+ "--master", "k8s://host:port",
+ "--class", "org.SomeClass",
+ "--conf", "spark.kubernetes.submitInDriver=true",
+ "--conf", "spark.kubernetes.archives.avoidDownloadSchemes=s3a",
+ "--conf",
"spark.hadoop.fs.s3a.impl=org.apache.spark.deploy.TestFileSystem",
+ "--conf", "spark.hadoop.fs.s3a.impl.disable.cache=true",
+ "--files", "src/test/resources/test_metrics_config.properties",
+ "--py-files", "src/test/resources/test_metrics_system.properties",
+ "--archives",
s"src/test/resources/log4j2.properties,$remoteArchiveFile",
+ "--jars", tmpJar.getAbsolutePath,
+ "/home/jarToIgnore.jar",
+ "arg1")
+ val appArgs = new SparkSubmitArguments(clArgs)
+ val (_, _, conf, _) = submit.prepareSubmitEnvironment(appArgs,
Some(hadoopConf))
+ conf.get("spark.master") should be("k8s://https://host:port";)
+ conf.get("spark.archives").contains(remoteArchiveFile) shouldBe true
+ conf.get("spark.archives").contains("log4j2.properties") shouldBe true
+
+ Files.exists(Paths.get("test_metrics_config.properties")) should be(true)
+ Files.exists(Paths.get("test_metrics_system.properties")) should be(true)
+ Files.exists(Paths.get("log4j2.properties")) should be(true)
+ Files.exists(Paths.get(tmpJar.getName)) should be(true)
+ Files.exists(Paths.get(notToDownloadArchive.getName)) should be(false)
+ Files.delete(Paths.get("test_metrics_config.properties"))
Review Comment:
These files are created in the current working directory (`core/`), but they
are not cleaned up when any assertion above fails. Could you do the cleanup in
`try { ... } finally { ... }`? Otherwise, the leftover files can affect the
subsequent test runs.
##########
core/src/main/scala/org/apache/spark/internal/config/package.scala:
##########
@@ -1632,6 +1632,19 @@ package object config {
.toSequence
.createWithDefault(Nil)
+ private[spark] val KUBERNETES_ARCHIVES_AVOID_DOWNLOAD_SCHEMES =
+ ConfigBuilder("spark.kubernetes.archives.avoidDownloadSchemes")
+ .doc("Comma-separated list of schemes for which archives will NOT be
downloaded to the " +
Review Comment:
This description is inaccurate for archives. Unlike jars
(`SparkContext.addJar` only registers the URI), `SparkContext.addFile(...,
isArchive = true)` always fetches the remote archive and unpacks it on the
driver. So, the archives are still downloaded to the driver local disk. What
this option avoids is the download in `SparkSubmit` and the executors fetching
the archives from the driver.
In addition,
- This takes effect only in K8s cluster mode (inside the driver pod), not
for all Kubernetes deployments.
- There are grammar issues copied from
`spark.kubernetes.jars.avoidDownloadSchemes`: `prior to be distributed` and `is
denoted to not downloading archives for any the schemes`.
The same applies to `docs/running-on-kubernetes.md`.
##########
core/src/test/scala/org/apache/spark/deploy/SparkSubmitSuite.scala:
##########
@@ -605,6 +605,109 @@ class SparkSubmitSuite
}
}
+ test("SPARK-55077: Avoid archives download if scheme matches " +
+ "spark.kubernetes.archives.avoidDownloadSchemes " +
+ "in k8s client mode & driver runs inside a POD") {
+ val hadoopConf = new Configuration()
+ updateConfWithFakeS3Fs(hadoopConf)
+ withTempDir { tmpDir =>
+ val notToDownloadArchive = File.createTempFile("NotToDownloadArchive",
".zip", tmpDir)
+ val remoteArchiveFile = s"s3a://${notToDownloadArchive.getAbsolutePath}"
+ val tmpJar = File.createTempFile("TestUDTF", ".jar", tmpDir)
+
+ val clArgs = Seq(
+ "--deploy-mode", "client",
+ "--proxy-user", "test.user",
+ "--master", "k8s://host:port",
+ "--class", "org.SomeClass",
+ "--conf", "spark.kubernetes.submitInDriver=true",
+ "--conf", "spark.kubernetes.archives.avoidDownloadSchemes=s3a",
+ "--conf",
"spark.hadoop.fs.s3a.impl=org.apache.spark.deploy.TestFileSystem",
+ "--conf", "spark.hadoop.fs.s3a.impl.disable.cache=true",
+ "--files", "src/test/resources/test_metrics_config.properties",
+ "--py-files", "src/test/resources/test_metrics_system.properties",
+ "--archives",
s"src/test/resources/log4j2.properties,$remoteArchiveFile",
+ "--jars", tmpJar.getAbsolutePath,
+ "/home/jarToIgnore.jar",
+ "arg1")
+ val appArgs = new SparkSubmitArguments(clArgs)
+ val (_, _, conf, _) = submit.prepareSubmitEnvironment(appArgs,
Some(hadoopConf))
+ conf.get("spark.master") should be("k8s://https://host:port";)
+ conf.get("spark.archives").contains(remoteArchiveFile) shouldBe true
+ conf.get("spark.archives").contains("log4j2.properties") shouldBe true
+
+ Files.exists(Paths.get("test_metrics_config.properties")) should be(true)
+ Files.exists(Paths.get("test_metrics_system.properties")) should be(true)
+ Files.exists(Paths.get("log4j2.properties")) should be(true)
+ Files.exists(Paths.get(tmpJar.getName)) should be(true)
+ Files.exists(Paths.get(notToDownloadArchive.getName)) should be(false)
+ Files.delete(Paths.get("test_metrics_config.properties"))
+ Files.delete(Paths.get("test_metrics_system.properties"))
+ Files.delete(Paths.get("log4j2.properties"))
+ Files.delete(Paths.get(tmpJar.getName))
+ }
+ }
+
+ test("SPARK-55077: Avoid archives download for wildcard scheme " +
+ "spark.kubernetes.archives.avoidDownloadSchemes " +
+ "in k8s client mode & driver runs inside a POD") {
+ val hadoopConf = new Configuration()
+ updateConfWithFakeS3Fs(hadoopConf)
+ withTempDir { tmpDir =>
+ val notToDownloadArchive =
File.createTempFile("NotToDownloadArchiveWildcard", ".zip", tmpDir)
+ val remoteArchiveFile = s"s3a://${notToDownloadArchive.getAbsolutePath}"
+
+ val clArgs = Seq(
+ "--deploy-mode", "client",
+ "--proxy-user", "test.user",
+ "--master", "k8s://host:port",
+ "--class", "org.SomeClass",
+ "--conf", "spark.kubernetes.submitInDriver=true",
+ "--conf", "spark.kubernetes.archives.avoidDownloadSchemes=*",
+ "--conf",
"spark.hadoop.fs.s3a.impl=org.apache.spark.deploy.TestFileSystem",
+ "--conf", "spark.hadoop.fs.s3a.impl.disable.cache=true",
+ "--archives", remoteArchiveFile,
+ "/home/jarToIgnore.jar",
+ "arg1")
+ val appArgs = new SparkSubmitArguments(clArgs)
+ val (_, _, conf, _) = submit.prepareSubmitEnvironment(appArgs,
Some(hadoopConf))
+ conf.get("spark.master") should be("k8s://https://host:port";)
+ conf.get("spark.archives").contains(remoteArchiveFile) shouldBe true
+ Files.exists(Paths.get(notToDownloadArchive.getName)) should be(false)
+ }
+ }
+
+ test("SPARK-55077: Download archives if scheme does not match " +
+ "spark.kubernetes.archives.avoidDownloadSchemes " +
+ "in k8s client mode & driver runs inside a POD") {
+ val hadoopConf = new Configuration()
+ updateConfWithFakeS3Fs(hadoopConf)
+ withTempDir { tmpDir =>
+ val toDownloadArchive = File.createTempFile("ToDownloadArchive",
".properties", tmpDir)
Review Comment:
Since this is not an archive, `Utils.unpack` only copies it with a warning,
so the actual unpacking is not tested. Could you use a real archive (e.g.
`.zip`) and cover the fragment form like `s3a://.../x.zip#alias` too, which is
the most common usage of archives? For the avoided case, `spark.archives`
should keep `#alias`. For the non-avoided case, `./alias` should be created.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
