Add run --spark-standalone

Author: alexarchambault

modules/cli-options/src/main/scala/scala/cli/commands/RunOptions.scala

@@ -30,6 +30,10 @@ final case class RunOptions(
   @ExtraName("spark")
     sparkSubmit: Option[Boolean] = None,
   @Group("Run")
+  @HelpMessage("Run as a Spark job, using a vanilla Spark distribution downloaded by Scala CLI")
+  @ExtraName("sparkStandalone")
+    standaloneSpark: Option[Boolean] = None,
+  @Group("Run")
   @HelpMessage("Print the command that would have been run (one argument per line), rather than running it")
     command: Boolean = false,
   @Group("Run")

modules/cli/src/main/scala/scala/cli/commands/Run.scala

@@ -25,7 +25,9 @@ object Run extends ScalaCommand[RunOptions] {
   override def sharedOptions(options: RunOptions): Option[SharedOptions] = Some(options.shared)
 
   private def runMode(options: RunOptions): RunMode =
-    if (options.sparkSubmit.getOrElse(false))
+    if (options.standaloneSpark.getOrElse(false) && !options.sparkSubmit.contains(false))
+      RunMode.StandaloneSparkSubmit
+    else if (options.sparkSubmit.getOrElse(false))
       RunMode.SparkSubmit
     else
       RunMode.Default
@@ -59,24 +61,24 @@ object Run extends ScalaCommand[RunOptions] {
             sharedJava.allJavaOpts.map(JavaOpt(_)).map(Positioned.commandLine),
         jvmIdOpt = baseOptions.javaOptions.jvmIdOpt.orElse {
           runMode(options) match {
-            case RunMode.SparkSubmit => Some("8")
-            case RunMode.Default     => None
+            case RunMode.StandaloneSparkSubmit | RunMode.SparkSubmit => Some("8")
+            case RunMode.Default                                     => None
           }
         }
       ),
       internalDependencies = baseOptions.internalDependencies.copy(
         addRunnerDependencyOpt = baseOptions.internalDependencies.addRunnerDependencyOpt.orElse {
           runMode(options) match {
-            case RunMode.SparkSubmit => Some(false)
-            case RunMode.Default     => None
+            case RunMode.StandaloneSparkSubmit | RunMode.SparkSubmit => Some(false)
+            case RunMode.Default                                     => None
           }
         }
       ),
       internal = baseOptions.internal.copy(
         keepResolution = baseOptions.internal.keepResolution || {
           runMode(options) match {
-            case RunMode.SparkSubmit => true
-            case RunMode.Default     => false
+            case RunMode.StandaloneSparkSubmit | RunMode.SparkSubmit => true
+            case RunMode.Default                                     => false
           }
         }
       ),
@@ -409,6 +411,18 @@ object Run extends ScalaCommand[RunOptions] {
                 scratchDirOpt
               )
             }
+          case RunMode.StandaloneSparkSubmit =>
+            value {
+              RunSpark.runStandalone(
+                build,
+                mainClass,
+                args,
+                logger,
+                allowExecve,
+                showCommand,
+                scratchDirOpt
+              )
+            }
         }
     }
   }

modules/cli/src/main/scala/scala/cli/commands/run/RunMode.scala

@@ -3,6 +3,7 @@ package scala.cli.commands.run
 sealed abstract class RunMode extends Product with Serializable
 
 object RunMode {
-  case object Default     extends RunMode
-  case object SparkSubmit extends RunMode
+  case object Default               extends RunMode
+  case object SparkSubmit           extends RunMode
+  case object StandaloneSparkSubmit extends RunMode
 }

modules/cli/src/main/scala/scala/cli/commands/util/RunSpark.scala

@@ -73,4 +73,74 @@ object RunSpark {
     }
   }
 
+  def runStandalone(
+    build: Build.Successful,
+    mainClass: String,
+    args: Seq[String],
+    logger: Logger,
+    allowExecve: Boolean,
+    showCommand: Boolean,
+    scratchDirOpt: Option[os.Path]
+  ): Either[BuildException, Either[Seq[String], (Process, Option[() => Unit])]] = either {
+
+    // FIXME Get Spark.sparkModules via provided settings?
+    val providedModules = Spark.sparkModules
+    val sparkClassPath  = value(PackageCmd.providedFiles(build, providedModules, logger))
+
+    scratchDirOpt.foreach(os.makeDir.all(_))
+    val library = os.temp(
+      Library.libraryJar(build),
+      dir = scratchDirOpt.orNull,
+      deleteOnExit = scratchDirOpt.isEmpty,
+      prefix = "spark-job",
+      suffix = ".jar"
+    )
+
+    val finalMainClass = "org.apache.spark.deploy.SparkSubmit"
+    val depCp          = build.dependencyClassPath.filterNot(sparkClassPath.toSet)
+    val javaHomeInfo   = build.options.javaHome().value
+    val javaOpts       = build.options.javaOptions.javaOpts.toSeq.map(_.value.value)
+    val jarsArgs =
+      if (depCp.isEmpty) Nil
+      else Seq("--jars", depCp.mkString(","))
+    val finalArgs =
+      Seq("--class", mainClass) ++
+        jarsArgs ++
+        javaOpts.flatMap(opt => Seq("--driver-java-options", opt)) ++
+        Seq(library.toString) ++
+        args
+    val envUpdates = javaHomeInfo.envUpdates(sys.env)
+    if (showCommand) {
+      val command = Runner.jvmCommand(
+        javaHomeInfo.javaCommand,
+        javaOpts,
+        sparkClassPath,
+        finalMainClass,
+        finalArgs,
+        extraEnv = envUpdates,
+        useManifest = build.options.notForBloopOptions.runWithManifest,
+        scratchDirOpt = scratchDirOpt
+      )
+      Left(command)
+    }
+    else {
+      val proc = Runner.runJvm(
+        javaHomeInfo.javaCommand,
+        javaOpts,
+        sparkClassPath,
+        finalMainClass,
+        finalArgs,
+        logger,
+        allowExecve = allowExecve,
+        extraEnv = envUpdates,
+        useManifest = build.options.notForBloopOptions.runWithManifest,
+        scratchDirOpt = scratchDirOpt
+      )
+      Right((
+        proc,
+        if (scratchDirOpt.isEmpty) Some(() => os.remove(library, checkExists = true))
+        else None
+      ))
+    }
+  }
 }

modules/integration/src/test/scala/scala/cli/integration/SparkTests212.scala

@@ -154,6 +154,18 @@ class SparkTests212 extends SparkTestDefinitions {
       expect(output.contains(expectedOutput))
     }
 
+  def simpleRunStandaloneSparkJobTest(spark: Spark): Unit =
+    simpleJobInputs(spark).fromRoot { root =>
+      val res = os.proc(TestUtil.cli, "run", extraOptions, "--spark-standalone", "--jvm", "8", ".")
+        .call(cwd = root)
+
+      val expectedOutput = "Result: 55"
+
+      val output = res.out.trim().linesIterator.toVector
+
+      expect(output.contains(expectedOutput))
+    }
+
   test("package spark 2.4") {
     simplePackageSparkJobTest(spark24)
   }
@@ -174,4 +186,12 @@ class SparkTests212 extends SparkTestDefinitions {
     simpleRunSparkJobTest(spark30, usePath = true)
   }
 
+  test("run spark 2.4 standalone") {
+    simpleRunStandaloneSparkJobTest(spark24)
+  }
+
+  test("run spark 3.0 standalone") {
+    simpleRunStandaloneSparkJobTest(spark30)
+  }
+
 }

website/docs/reference/cli-options.md

@@ -1476,6 +1476,12 @@ Aliases: `--spark`
 
 Run as a Spark job, using the spark-submit command
 
+#### `--standalone-spark`
+
+Aliases: `--spark-standalone`
+
+Run as a Spark job, using a vanilla Spark distribution downloaded by Scala CLI
+
 #### `--command`
 
 Print the command that would have been run (one argument per line), rather than running it

website/src/pages/spark.md

@@ -63,3 +63,13 @@ scala-cli run --spark SparkJob.scala # same example as above
 Note that this requires either
 - `spark-submit` to be in available in `PATH`
 - `SPARK_HOME` to be set in the environment
+
+## Running Spark jobs in a standalone way
+
+The `run` sub-command can not only run Spark jobs, but it can also work without a Spark
+distribution. For that to work, it downloads Spark JARs, and calls the main class of
+`spark-submit` itself via these JARs:
+
+```bash
+scala-cli run --spark-standalone SparkJob.scala # same example as above
+```