org.apache.spark.SparkException: Tâche de ne pas sérialisable

Ce travail est une exemple de code: 

JavaPairDStream<String, String> messages = KafkaUtils.createStream(javaStreamingContext, zkQuorum, group, topicMap);
messages.print();
JavaDStream<String> lines = messages.map(new Function<Tuple2<String, String>, String>() {
    @Override
    public String call(Tuple2<String, String> tuple2) {
        return tuple2._2();
    }
});

J'obtiens l'erreur ci-dessous:

ERROR:
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:1435)
    at org.apache.spark.streaming.dstream.DStream.map(DStream.scala:438)
    at org.apache.spark.streaming.api.java.JavaDStreamLike$class.map(JavaDStreamLike.scala:140)
    at org.apache.spark.streaming.api.java.JavaPairDStream.map(JavaPairDStream.scala:46)
  • Eh bien, si ça fonctionne c'est génial :). Si elle ne l'est pas, vous pouvez activer la sérialisation Java débogage avec -Dsun.io.serialization.extendedDebugInfo=true.
  • Merci,c'est pas ok,j'ai essayé. JavaDStream<String> lignes = messages.carte(la nouvelle Fonction<Tuple2<String, String>, String>() { @Override public String appel(Tuple2<String, String> tuple2) { return tuple2._2(); } }); Cette ligne de code de problèmes.
  • Assez sûr que ce code est en Java et non Scala (c'est à dire de la balise)
InformationsquelleAutor xiaolong li | 2015-03-27
  1. 15

    Puisque vous êtes à la définition de votre fonction map à l'aide d'un anonyme intérieur de la classe, la classe doit être Sérialisable. Définissez votre carte de fonction en tant que catégorie distincte ou d'en faire un statique à l'intérieur de la classe. À partir de la documentation de Java (http://docs.oracle.com/javase/8/docs/platform/serialization/spec/serial-arch.html):

    Remarque - la Sérialisation des classes internes (c'est à dire, classes imbriquées qui ne sont pas membre statique des classes), y compris les locaux et anonyme des classes, est fortement déconseillée pour plusieurs raisons. Parce que les classes internes déclarés dans les contextes non statiques contiennent implicite de non-transitoire références à la classe englobante cas, la sérialisation d'une telle intérieure instance de classe sera le résultat de la sérialisation de ses associés extérieur instance de classe ainsi.

    • Merci beaucoup！
    • Content que ça a aidé! Veuillez accepter la réponse et si elle n'
    • La Spark est d'essayer de sérialiser l'objet transmis à la carte, mais ne peut pas sérialiser parce que ne pas Sérialisable? Pourquoi Étincelle est en train de faire de la sérialisation? Et si nous définissons la fonction map comme une classe distincte, n'avons-nous besoin de le faire Serializable ainsi?
    InformationsquelleAutor InPursuit
  2. 4

    simplement de donner l'exemple de code :

    JavaDStream<String> lines = messages.map(mapFunc);

    déclarer l'intérieur de la classe comme une variable statique :

    static Function<Tuple2<String, String>, String> mapFunc=new Function<Tuple2<String, String>, String>() {
    @Override
    public String call(Tuple2<String, String> tuple2) {
        return tuple2._2();
    }
}
    • Ses résolu mon problème. Merci
    InformationsquelleAutor udyan