Spark Scala Dynamische Erstellung von Serializable-Objekt

Question

Ich brauche einen Tester für Scala Spark-Filter, mit Tester Java-Prädikat-Schnittstelle implementieren und bestimmten Klassennamen durch Argumente erhalten. Ich ist

val tester = Class.forName(qualifiedName).newInstance().asInstanceOf[Predicate[T]] 
var filtered = rdd.filter(elem => tester.test(elem)) 

Das Problem wie dieses, etwas zu tun, dass ich zur Laufzeit eine Spark „TaskNotSerializable Exception“, weil meine spezifischen Prädikatsklasse haben, ist nicht Serializable.

Wenn ich

val tester = Class.forName(qualifiedName).newInstance() 
      .asInstanceOf[Predicate[T] with Serializable] 
var filtered = rdd.filter(elem => tester.test(elem)) 

Ich bekomme den gleichen Fehler. Wenn ich Tester in rdd.filter erstellen nennen es funktioniert:

var filtered = rdd.filter { elem => 
    val tester = Class.forName(qualifiedName).newInstance() 
      .asInstanceOf[Predicate[T] with Serializable] 
    tester.test(elem) 
} 

Aber ich würde ein einzelnes Objekt (vielleicht auszustrahlen) zum Testen erstellen. Wie kann ich auflösen?

Answer 1

Sie müssen nur die Klasse implementieren Serializable. Beachten Sie, dass die asInstanceOf[Predicate[T] with Serializable] Umwandlung eine Lüge ist: Sie prüft nicht, ob der Wert Serializable ist, weshalb der zweite Fall nicht sofort während der Umwandlung einen Fehler erzeugt und der letzte "erfolgreich" ist.

Aber ich würde ein einzelnes Objekt (vielleicht zu Broadcast) zum Testen erstellen.

Sie können nicht. Broadcast oder nicht, Deserialisierung erstellt neue Objekte auf Worker-Knoten. Sie können jedoch nur eine einzige Instanz auf jeder Partition zu erstellen:

var filtered = rdd.mapPartitions { iter => 
    val tester = Class.forName(qualifiedName).newInstance() 
      .asInstanceOf[Predicate[T]] 
    iter.filter(tester.test) 
} 

Es wird tatsächlich eine bessere Leistung als die tester Serialisierung, sendet es und Deserialisieren wäre es, da es streng weniger Arbeit.