Wie senden Sie einen Job von Code an Spark?

Question

Ich habe einen eigenständigen Single-Node-Spark-Cluster erstellt und bestätigt, dass mein Build mit ./bin/run-example SparkPi 10 funktioniert. Dann habe ich ein wirklich einfaches Testprojekt in scala geschrieben;

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) 
    } 
} 

Ich laufe dies aus meinem IDE (IntelliJ). Es verbindet sich erfolgreich mit dem Cluster, und ich sehe, dass es Jobs übermittelt, aber alle werfen einen Fehler auf;

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] 

Wenn ich Funke richtig verstehe, ist dies, weil der Cluster den Code nicht finden kann. Wie füttere ich den Code, um zu funken? Ich führe HDFS oder irgendetwas in diesem Test nicht durch, aber es ist alles auf einem Kasten, also hätte ich SparkContext erwartet, um das gegenwärtige Verzeichnis zu übergeben, aber es tut anscheinend nicht.

Kann mir jemand auf den richtigen Weg zeigen, dies einzurichten?

Answer 1

Wenn Sie Ihr Spark-Programm lokal testen möchten, müssen Sie den Einzelknoten-Spark nicht einmal hochfahren. Geben Sie einfach Ihre Master-URL zu local[*] wie diese

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

Dann in sbt, Typ > run Ihr Programm auszuführen (dies soll das gleiche sein wie Laufen von IntelliJ, aber ich verwenden, um das Programm von Terminal sbt laufen).

Da Sie nicht zwischen local[*] und spark://... oft im Code können Ihre Master-URL ändern, können Sie sie leer lassen

val sc = new SparkContext(new SparkConf()) 

und Ihre Java-Eigenschaften festgelegt, wenn ausgeführt wird, zum Beispiel in build.sbt, Sie kann

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

hinzufügen und es laufen run in sbt verwenden.

---

Um einen umfassenderen lokalen Modus Erfahrung zu machen, sollten Sie die folgenden Zeilen in Ihrer 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) 

Wir haben ein sbt Plugin erstellt hinzuzufügen, die für diese Einstellungen hinzufügen können, es kann Ihnen auch helfen, einen eigenständigen Spark-Cluster auf einem Cloud-System wie aws ec2 zu implementieren, werfen Sie einen Blick auf spark-deployer, wenn Sie interessiert sind.

Answer 2

Sie sind ein entscheidender Schritt fehlt:

org.apache.spark.deploy.SparkSubmit

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

, die tatsächlich einen Auftrag an den Cluster einreicht. Leider gibt es zur Zeit keine solide Arbeitshülle für ihn außer spark-submit. Daher gibt es derzeit keinen zuverlässigen Weg, um Funkenjobs programmatisch einzureichen. Es gibt eine Jira dafür, die im Februar 2015 teilweise angesprochen wurde: aber es fehlt Dokumentation.

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

Die Schwierigkeit liegt in der Komplexität der Umweltmachenschaften von spark-submit zur Verfügung gestellt. Es wurde nicht gefunden, dass sie nur im Scala/Java-Code repliziert werden können.

Answer 3

Hier ist die Klasse.

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(); 
    } 
    }