Verschlechtert die Serialisierung die Spark-Leistung?

Question

die folgende Fall Klasse Gegeben:

case class User(name:String, age:Int) 

Ein RDD von einem List von Instanzen Benutzer

Der folgende Code erstellt wird, um die RDD filtert Benutzer von 50 über das Alter entfernen

trait Process { 
    def test { 
    val rdd = ... // create RDD 
    rdd.filter(_.age>50) 
    } 
} 

Um die Protokollierung hinzuzufügen, wird eine separate Überprüfungsfunktion erstellt und wie folgt an den Filter übergeben:

trait Process { 
    def validate(user:User) { 
    if (user.age>50) { 
     true 
    } 
    else { 
     println("FAILED VALIDATION") 
     false 
    } 
    } 

    def test { 
    val rdd = ... // create RDD 
    rdd.filter(validate) 
    } 
} 

Die folgende Ausnahme ausgelöst:

org.apache.spark.SparkException: Task not serializable 

Der Code, indem die Klasse arbeitet, in dem die Prüf-Funktion ist serializable definiert:

trait Process extends Serializable 

Ist dies der richtige Weg, um die Task not serializable zu handhaben Ausnahme oder gibt es eine Leistungseinbuße bei der Verwendung der Serialisierung in Spark? Gibt es bessere Möglichkeiten, dies zu tun?

Dank

Answer 1

gibt es eine Leistungsverschlechterung der Serialisierung innerhalb Funken

Aufgabe Serialisierung zu verwenden (im Gegensatz zu Daten Serialisierung im Gegensatz, die beim Shuffling/Sammeln von Daten auftreten) ist selten merklich leistungsmäßig, solange die serialisierten Objekte klein sind. Die Taskserialisierung erfolgt einmal pro Task (unabhängig von der Menge der verarbeiteten Daten).

In diesem Fall (Serialisierung der Process-Instanz) wäre die Auswirkung auf die Leistung wahrscheinlich vernachlässigbar, da es sich um ein kleines Objekt handelt.

Das Risiko mit dieser Annahme („Process ist klein, so dass es in Ordnung ist“) ist die Zeit vorbei, Process ändern könnte: es wäre einfach für Entwickler wird nicht bemerken, dass diese Klasse serialisiert, so könnten sie hinzufügen Mitglieder, die das langsamer machen würden.

Gibt es bessere Möglichkeiten, diese

Sie Serialisierung vollständig unter Verwendung von statischen Methoden vermeiden können, zu tun - Methoden der objects statt Klassen.

import Process._ 

trait Process { 
    def test { 
    val rdd = ... // create RDD 
    rdd.filter(validate) 
    } 
} 

object Process { 
    def validate(user:User) { 
    if (user.age>50) { 
     true 
    } else { 
     println("FAILED VALIDATION") 
    false 
    } 
    } 

Objekte sind „statisch“, so Funke kann sie ohne Serialisierung verwenden: In diesem Fall können Sie ein Begleiter Objekt für Process erstellen.