Gibt es einen Haken für Executor Startup in Spark?

Question

Also im Grunde möchte ich mehrere Aufgaben auf dem gleichen Knoten/Executor zum Lesen von Daten aus einem gemeinsamen Speicher ausführen. Dazu brauche ich eine Initialisierungsfunktion, die die Daten vor dem Start der Tasks in den Speicher lädt. Wenn Spark einen Hook für einen Executor-Start bereitstellt, könnte ich diesen Initialisierungscode in diese Rückruffunktion einfügen, wobei die Aufgaben nur ausgeführt werden, nachdem dieser Startvorgang abgeschlossen ist.

Also, meine Frage ist, bietet Spark solche Haken? Wenn nicht, mit welcher anderen Methode kann ich dasselbe erreichen?

Answer 1

Sie müssen nicht mehrere Instanzen der App ausführen, um mehrere Tasks ausführen zu können (z. B. eine Anwendungsinstanz, eine Spark-Task). Das gleiche SparkSession-Objekt kann von mehreren Threads verwendet werden, um Spark-Aufgaben parallel zu übergeben.

So kann es wie folgt funktionieren:

• Die Anwendung startet und betreibt eine Initialisierungsfunktion gemeinsam genutzte Daten in den Speicher zu laden. Sprich, in ein SharedData-Klassenobjekt.
• SparkSession wird
erstellt
• A Threadpool erstellt wird, wobei jeder Thread Zugriff auf (SparkSession, Shared) Objekte
• Jeder Thread erstellt Funken Aufgabe Verwendung gemeinsamer SparkSession und Shared Objekte hat.
• Je nach Anwendungsfall die Anwendung tut dann eine der folgenden:
  • wartet für alle Aufgaben zu erledigen und dann Session
  • wartet in einer Schleife schließt Funken für neue Anforderungen und schafft neue Spark ankommen Aufgaben wie erforderlich mit Threads aus dem Thread-Pool.

SparkContext (sparkSession.sparkContext) ist nützlich, wenn Sie pro Thread Dinge tun wollen wie eine Aufgabenbeschreibung Zuweisen einer Gruppe, um die Aufgabe setJobGroup so damit verbundenen Aufgaben können gleichzeitig abgebrochen mit setJobDescription oder Zuweisung verwenden cancelJobGroup. Sie können auch die Priorität für die Aufgaben optimieren, die den gleichen Pool verwenden, siehe https://spark.apache.org/docs/latest/job-scheduling.html#scheduling-within-an-application für Details.

Answer 2

Spark Lösung für „shared data“ wird mit Broadcast - wo Sie die Daten laden einmal im Treiber Anwendung und Spark-serialisiert und sendet an jeden der Testamentsvollstrecker (einmal). Wenn eine Aufgabe diese Daten verwendet, stellt Spark sicher, dass sie vorhanden ist, bevor die Aufgabe ausgeführt wird. Zum Beispiel:

object MySparkTransformation { 

    def transform(rdd: RDD[String], sc: SparkContext): RDD[Int] = { 
    val mySharedData: Map[String, Int] = loadDataOnce() 
    val broadcast = sc.broadcast(mySharedData) 
    rdd.map(r => broadcast.value(r)) 
    } 
} 

Alternativ, wenn Sie das Lesen der Daten in den Treiber Speicher vermeiden wollen und es über den Testamentsvollstrecker senden, können Sie lazy Werte in einer Scala object verwenden, um einen Wert zu erzeugen, die einmal bevölkert wird pro JVM, die in Sparks Fall einmal pro Executor ist. Zum Beispiel:

// must be an object, otherwise will be serialized and sent from driver 
object MySharedResource { 
    lazy val mySharedData: Map[String, Int] = loadDataOnce() 
} 

// If you use mySharedData in a Spark transformation, 
// the "local" copy in each executor will be used: 
object MySparkTransformation { 
    def transform(rdd: RDD[String]): RDD[Int] = { 
    // Spark won't include MySharedResource.mySharedData in the 
    // serialized task sent from driver, since it's "static" 
    rdd.map(r => MySharedResource.mySharedData(r)) 
    } 
} 

In der Praxis werden Sie eine Kopie mySharedData in jedem Testamentsvollstrecker haben.