[K8S] Extend avoidDownloadSchemes to spark.files so executors can read uploaded files directly

[WonYong-Jang]

Summary

In K8s cluster mode, SparkSubmit unconditionally downloads spark.files to the driver's local disk and rewrites the URIs to file:/tmp/spark-<uuid>/..., discarding the original remote URI. (https://issues.apache.org/jira/browse/SPARK-47475) already made this skippable for jars via spark.kubernetes.jars.avoidDownloadSchemes (4.0.0). This asks for the same for files.

Current behavior

--files uploads to spark.kubernetes.file.upload.path, but the driver then localizes it and the s3a:// URI is lost before user code runs:

// core/src/main/scala/org/apache/spark/deploy/SparkSubmit.scala
if (isKubernetesClusterModeDriver) {
  val filesLocalFiles = Option(args.files).map {
    downloadResourcesToCurrentDirectory(_)                                   // always downloaded
  }.orNull
  val updatedJars = Option(args.jars).map {
    downloadResourcesToCurrentDirectory(_, avoidDownload = avoidJarDownload) // skippable by scheme
  }.orNull
  args.files = filesLocalFiles
}

spark.sparkContext.getConf().get("spark.files")
# expected: s3a://bucket/spark-upload-<uuid>/test.csv
# actual:   file:/tmp/spark-fff175ac-.../test.csv

This is scheme-independent, so passing --files s3a://bucket/path/test.csv is localized too.

Why the URI matters

To be clear about what is and isn't working: the files are delivered to executors.
addFile registers them with the driver's file server, Executor.updateDependencies fetches them on startup, and SparkFiles.get("test.csv") returns a valid path inside a task.

The problem is that spark.read.* needs something different - one path string, resolved once on the driver, that every executor can open. No such string exists here:

• The driver-local path(/tmp/spark-<uuid>/test.csv) does not exist on executors.
• SparkFiles.get() returns a different path depending on where it is called - a driver temp dir on the driver, the container working directory on executors - so a path resolved on the driver and shipped to executors points at nothing.

The s3a:// upload path is the one address that would work everywhere, and Spark already computes it. It is just discarded before user code can see it.

Proposal

Apply the existing avoidDownload predicate to files:

val avoidFileDownloadSchemes = sparkConf.get(KUBERNETES_FILES_AVOID_DOWNLOAD_SCHEMES)

def avoidFileDownload(scheme: String): Boolean =
  avoidFileDownloadSchemes.contains("*") || avoidFileDownloadSchemes.contains(scheme)

val filesLocalFiles = Option(args.files).map {
  downloadResourcesToCurrentDirectory(_, avoidDownload = avoidFileDownload)
}.orNull

SparkContext never rewrites spark.files, so with spark.kubernetes.files.avoidDownloadSchemes=s3a the remote URI reaches user code, and addFile keeps the s3a key so executors fetch from object storage instead of the driver file server - the same driver-network relief that motivated (https://issues.apache.org/jira/browse/SPARK-47475).

Reproduction

spark-submit --master k8s://<api-server> --deploy-mode cluster \
  --files test.csv \
  --conf spark.kubernetes.file.upload.path=s3a://bucket/uploads \
  app.py

print(spark.sparkContext.getConf().get("spark.files"))  # s3a URI is gone

Current workaround

with open(SparkFiles.get("test.csv"), encoding="utf-8") as f:
    rdd = spark.sparkContext.parallelize(f.read().splitlines())
df = spark.read.csv(rdd, header=True)

Routes the whole file through driver memory; doesn't scale.