java+spark: org.apache.spark.SparkException: Job abgebrochen: Aufgabe nicht serialisierbar: java.io.NotSerializableException

Ich bin neu zu entfachen, und versuchte, um das Beispiel auszuführen JavaSparkPi.java es läuft auch gut, aber da ich dies in einem anderen java-s kopiere ich alle Sachen vom main auf eine Methode in der Klasse und versuchen, rufen Sie die Methode in main, es saids

org.apache.spark.SparkException: Job abgebrochen: Aufgabe nicht serialisierbar:
java.io.NotSerializableException

der code sieht so aus:

public class JavaSparkPi {

public void cal(){
    JavaSparkContext jsc = new JavaSparkContext("local", "JavaLogQuery");
    int slices = 2;
    int n = 100000 * slices;

    List<Integer> l = new ArrayList<Integer>(n);
    for (int i = 0; i < n; i++) {
        l.add(i);
    }

    JavaRDD<Integer> dataSet = jsc.parallelize(l, slices);

    System.out.println("count is: "+ dataSet.count());
    dataSet.foreach(new VoidFunction<Integer>(){
        public void call(Integer i){
            System.out.println(i);
        }
    });

    int count = dataSet.map(new Function<Integer, Integer>() {
        @Override
        public Integer call(Integer integer) throws Exception {
            double x = Math.random() * 2 - 1;
            double y = Math.random() * 2 - 1;
            return (x * x + y * y < 1) ? 1 : 0;
        }
    }).reduce(new Function2<Integer, Integer, Integer>() {
        @Override
        public Integer call(Integer integer, Integer integer2) throws Exception {
            return integer + integer2;
        }
    });

    System.out.println("Pi is roughly " + 4.0 * count / n);
}

public static void main(String[] args) throws Exception {

    JavaSparkPi myClass = new JavaSparkPi();
    myClass.cal();
}
}

jemand ne Idee hierzu? danke!

InformationsquelleAutor user2810081 | 2014-06-04
  1. 14

    Den verschachtelten Funktionen halten eine Referenz auf das enthaltene Objekt (JavaSparkPi). Also dieses Objekt wird serialisiert. Damit dies funktioniert, muss es serialisierbar sein. Ganz einfach:

    public class JavaSparkPi implements Serializable {
  ...
    • Das ist Recht, oder, um die anonyme innere Klassen, statische Klassen, wenn Sie nicht können oder nicht wollen, um die äußeren Klasse statisch. Das ist mehr Aufwand code-klug, aber
    • stackoverflow.com/questions/40818001/... eine übersicht über die für das Verständnis der spark-Serialisierung
    • Sie rocken Mann, für mich funktioniert wie ein Charme!
    InformationsquelleAutor Daniel Darabos
  2. 2

    Das Hauptproblem ist, dass, wenn Sie eine Anonyme Klasse in java übergeben wird eine Referenz auf die umschließende Klasse.
    Dies kann behoben werden, in vielerlei Hinsicht

    Erklären der einschließenden Klasse Serializable

    Das funktioniert in deinem Fall aber wird fallen flach, im Falle Ihrer umschließenden Klasse hat auch ein paar Felder, die nicht serialisierbar sind. Ich würde auch sagen, dass der Serialisierung der übergeordneten Klasse ist eine totale Verschwendung.

    Erstellen Schließung in eine statische Funktion

    Erstellen der Schließung durch aufrufen der statischen Funktion nicht, übergeben Sie die Referenz auf die Schließung und damit keine Notwendigkeit zu stellen Sie serializable auf diese Weise.

    InformationsquelleAutor Anuj J
  3. 0

    Dieser Fehler kommt, weil Sie mehrere physische CPUs und in Ihrem lokalen oder cluster-und spark-engine zu senden versuchen Sie diese Funktion, um mehrere CPUs über ein Netzwerk.
    Ihre Funktion

     dataSet.foreach(new VoidFunction<Integer>(){
        public void call(Integer i){
            ***System.out.println(i);***
        }
    });

    verwendet println (), die sich nicht serialisieren. Also die Ausnahme, die ausgelöst durch Spark-Motor.
    Die Lösung, die Sie verwenden können unten:

    dataSet.collect().forEach(new VoidFunction<Integer>(){
       public void call(Integer i){
         System.out.println(i);
    }
});
    InformationsquelleAutor Jeetu
Schreibe einen Kommentar