Aufgabe nicht Serialisierung Ausnahme beim Ausführen von Apache Funke Job

Question

Das folgende Java-Programm wurde geschrieben, um mit Apache Funke zu experimentieren.

Das Programm versucht, eine Liste positiver und negativer Wörter aus einer entsprechenden Datei zu lesen, diese mit der Hauptdatei zu vergleichen und die Ergebnisse entsprechend zu filtern.

import java.io.Serializable; 
import java.io.FileNotFoundException; 
import java.io.File; 
import java.util.*; 
import java.util.Iterator; 
import java.util.List; 
import java.util.List; 
import org.apache.spark.api.java.*; 
import org.apache.spark.api.java.function.Function; 

public class SimpleApp implements Serializable{ 
    public static void main(String[] args) { 
    String logFile = "/tmp/master.txt"; // Should be some file on your system 
    String positive = "/tmp/positive.txt"; // Should be some file on your system 
    String negative = "/tmp/negative.txt"; // Should be some file on your system 

    JavaSparkContext sc = new JavaSparkContext("local[4]", "Twitter Analyzer", "/home/welcome/Downloads/spark-1.1.0/", new String[]{"target/scala-2.10/Simple-assembly-0.1.0.jar"}); 

    JavaRDD<String> positiveComments = sc.textFile(logFile).cache(); 

    List<String> positiveList = GetSentiments(positive); 
    List<String> negativeList= GetSentiments(negative); 

    final Iterator<String> iterator = positiveList.iterator(); 
    int i = 0; 
    while (iterator.hasNext()) 
    { 
     JavaRDD<String> numAs = positiveComments.filter(new Function<String, Boolean>() 
     { 
     public Boolean call(String s) 
     { 
      return s.contains(iterator.next()); 
     } 
     }); 

    numAs.saveAsTextFile("/tmp/output/"+ i); 
    i++; 
    } 

    } 

public static List<String> GetSentiments(String fileName) { 
    List<String> input = new ArrayList<String>(); 
try 
{ 
    Scanner sc = new Scanner(new File(fileName)); 

    while (sc.hasNextLine()) { 
     input.add(sc.nextLine()); 
    } 
} 
catch (FileNotFoundException e){ 
    // do stuff here.. 
} 
    return input; 
} 

} 

Der folgende Fehler wird

Exception in thread "main" org.apache.spark.SparkException: Task not serializable 
    at org.apache.spark.util.ClosureCleaner$.ensureSerializable(ClosureCleaner.scala:166) 
    at org.apache.spark.util.ClosureCleaner$.clean(ClosureCleaner.scala:158) 
    at org.apache.spark.SparkContext.clean(SparkContext.scala:1242) 
    at org.apache.spark.rdd.RDD.filter(RDD.scala:282) 
    at org.apache.spark.api.java.JavaRDD.filter(JavaRDD.scala:78) 
    at SimpleApp.main(SimpleApp.java:37) 
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) 
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57) 
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) 
    at java.lang.reflect.Method.invoke(Method.java:606) 
    at org.apache.spark.deploy.SparkSubmit$.launch(SparkSubmit.scala:328) 
    at org.apache.spark.deploy.SparkSubmit$.main(SparkSubmit.scala:75) 
    at org.apache.spark.deploy.SparkSubmit.main(SparkSubmit.scala) 
Caused by: java.io.NotSerializableException: java.util.ArrayList$Itr 
    at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1183) 
    at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1547) 
    at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1508) 
    at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1431) 
    at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1177) 
    at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1547) 
    at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1508) 
    at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1431) 
    at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1177) 
    at java.io.ObjectOutputStream.writeObject(ObjectOutputStream.java:347) 
    at org.apache.spark.serializer.JavaSerializationStream.writeObject(JavaSerializer.scala:42) 
    at org.apache.spark.serializer.JavaSerializerInstance.serialize(JavaSerializer.scala:73) 
    at org.apache.spark.util.ClosureCleaner$.ensureSerializable(ClosureCleaner.scala:164) 
    ... 12 more 

Alle Zeiger während der Ausführung von Funken Job, geworfen ??

Answer 1

Wenn Sie eine anonyme Klasse erstellen, wird der Compiler einige Sachen:

JavaRDD<String> numAs = positiveComments.filter(new Function<String, Boolean>() 
     { 
     public Boolean call(String s) 
     { 
      return s.contains(iterator.next()); 
     } 
     }); 

Es wird wie folgt umgeschrieben werden:

JavaRDD<String> numAs = positiveComments.filter(new Function<String, Boolean>() 
     { 
     private Iterator<...> $iterator; 
     public Boolean call(String s) 
     { 
      return s.contains($iterator.next()); 
     } 
     }); 

diesem Grund sollten Sie eine NotSerializableException haben kann, weil der Iterator nicht ist serialisierbar.

Um das zu vermeiden, extrahieren einfach das Ergebnis der nächsten vor:

String value = iterator.next(); 
JavaRDD<String> numAs = positiveComments.filter(new Function<String, Boolean>() 
     { 
     public Boolean call(String s) 
     { 
      return s.contains(value); 
     } 
     }); 

Answer 2

Einige Java Fakten

1. Any anonymous class defined inside a outer class has reference to the outer class. 
    2. If the anonymous class needs to be serialized it will compel you to make the outer class serialized. 
    3. Inside the lambda function if one uses a method of the enclosing class , the class needs to be serialized , if the lambda function is being serialized. 

Einige Fakten über Spark.

1. On Same Executor multiple tasks can run at the same time in the same JVM as Tasks are spawned as threads in spark. 
2. Any lambda, Anonymous Class used with the spark Transformation function (map, mapPartitions, keyBy , redudeByKey …) will be instantiated on driver, serialized and sent to the executor. 
3. To serialize an object means to convert its state to a byte stream so that the byte stream can be reverted back into a copy of the object. 
4. A Java object is serializable if its class or any of its super class implements either the java.io.Serializable interface or its subinterface, java.io.Externalizable. 

Daumenregel Serialisierung Problem zu vermeiden:

1. Avoid using anonymous class , instead use static classes as anonymous class will force you to have the outer class serialized. 
    2. Avoid using static variables as a work around for serialization issue , as Multiple Task can run inside the same JVM and the static instance might not be thread safe. 
    3. Use Transient variables to avoid serialization issue , you will have to initialize them inside the function call and not Constructor. As on driver the constructor will be called , on Executor it will de-serialize and for the object . only way to initialize is inside the function call . 
    4. Use Static class in place of anonymous class. 
    5. Religiously follow ” attaching implements Serializable ” only for the classes which only needs to be serialized 
    6. Inside a “lambda function” never refer to outclass method directly , as this will lead to serialization of outer class. 
    7. Make methods static if it needs to be used within Lambda function directly , else use Class::func() notion but not func() directly 
    8. Java Map<> doesn’t implement Serializable but HashMap does . 
    9. Be wise when deciding over using Braodcast vs Raw DataStructures. If you see a real benefit then only use Broadcast. 

Für ein tieferes Verständnis folgen http://bytepadding.com/big-data/spark/understanding-spark-serialization/