Как отправить вакансию Spark из кода?

Question

Я объединил автономный Spark-кластер с одним узлом и подтвердил мои работы по строительству с помощью ./bin/run-example SparkPi 10. Затем я написал очень простой тестовый проект в scala;

import org.apache.spark.{SparkConf, SparkContext} 

object Main { 
    def main(args: Array[String]): Unit = { 
    val sparkConf = new SparkConf() 
    val sc = new SparkContext("spark://UbuntuVM:7077", "Simple Application") 

    val count = sc.parallelize(1 to 100).map{i => 
     val x = Math.random() 
     val y = Math.random() 
     if (x*x + y*y < 1) 1 else 0 
    }.reduce(_ + _) 
    println("Pi is roughly " + 4.0 * count/100) 
    } 
} 

Я запускаю это изнутри своей IDE (IntelliJ). Он успешно подключается к кластеру, и я вижу, что он отправляет задания, но все они выдают ошибку;

INFO TaskSetManager: Lost task 1.3 in stage 0.0 (TID 7) on executor 192.168.1.233: java.lang.ClassNotFoundException (Main$$anonfun$1) [duplicate 7] 

Если я правильно понял искру, это происходит потому, что кластер не может найти код. Итак, как я могу накормить код для искры? Я не запускаю HDFS или что-то в этом тесте, но все это на одной коробке, поэтому я бы ожидал, что SparkContext передаст текущую директорию в искру, но, похоже, этого не делает.

Может ли кто-нибудь указать мне правильный способ установить это?

Answer 1

Если вы хотите протестировать свою программу Spark локально, вам даже не нужно разворачивать одиночную узлу Spark. Просто установите мастер URL в local[*] как этот

val sc = new SparkContext("local[*]", "Simple Application", sparkConf) 

Тогда в СБТ, тип > run для запуска программы (это должно быть таким же, как работает с IntelliJ, но я использовал, чтобы запустить программу из SBT терминала с использованием).

Поскольку вы не можете изменить свой основной URL в коде между local[*] и spark://... много раз, вы можете оставить их пустыми

val sc = new SparkContext(new SparkConf()) 

и установить свойства Java при работе, например, в build.sbt, вы может добавить

javaOptions := Seq("-Dspark.master=local[*]", "-Dspark.app.name=my-app") 

и запустить его с помощью run в SBT.

---

Чтобы сделать более обширный опыт местного режима, вы можете добавить следующие строки в build.sbt

run in Compile <<= Defaults.runTask(fullClasspath in Compile, mainClass in (Compile, run), runner in (Compile, run)) 
runMain in Compile <<= Defaults.runTask(fullClasspath in Compile, mainClass in (Compile, run), runner in (Compile, run)) 
fork := true 
javaOptions := Seq("-Dspark.master=local[*]", s"-Dspark.app.name=my-app") 
outputStrategy := Some(StdoutOutput) 

Мы создали SBT плагин, который может добавить эти настройки для вас, это также может помочь вам развернуть автономный Spark-кластер в облачной системе, такой как aws ec2, посмотрите, пожалуйста, на spark-deployer, если вы заинтересованы.

Answer 2

Вы пропускаете важный шаг:

org.apache.spark.deploy.SparkSubmit

https://github.com/apache/spark/blob/master/core/src/main/scala/org/apache/spark/deploy/SparkSubmit.scala

, который на самом деле представляет работу в кластере. К сожалению, в настоящее время нет сплошной рабочей обертки, кроме spark-submit. Таким образом, в настоящее время это не надежный способ программно представить искровые задания. Для него есть джира, которая была частично решена в феврале 2015 года, но ей не хватает документации.

https://github.com/apache/spark/pull/3916/files 

Трудность заключается в сложности экологических махинаций, предусмотренных spark-submit. Невозможно воспроизвести их исключительно в коде scala/java.

Answer 3

Это класс.

https://spark.apache.org/docs/1.4.0/api/java/org/apache/spark/launcher/package-summary.html

import org.apache.spark.launcher.SparkLauncher; 

    public class MyLauncher { 
    public static void main(String[] args) throws Exception { 
     Process spark = new SparkLauncher() 
     .setAppResource("/my/app.jar") 
     .setMainClass("my.spark.app.Main") 
     .setMaster("local") 
     .setConf(SparkLauncher.DRIVER_MEMORY, "2g") 
     .launch(); 
     spark.waitFor(); 
    } 
    }