You are viewing a plain text version of this content. The canonical link for it is here.
Posted to user@spark.apache.org by Spark Enthusiast <sp...@yahoo.in> on 2015/06/29 14:14:54 UTC
Serialization Exception
For prototyping purposes, I created a test program injecting dependancies using Spring.
Nothing fancy. This is just a re-write of KafkaDirectWordCount. When I run this, I get the following exception:
Exception in thread "main" org.apache.spark.SparkException: Task not serializable
at org.apache.spark.util.ClosureCleaner$.ensureSerializable(ClosureCleaner.scala:315)
at org.apache.spark.util.ClosureCleaner$.org$apache$spark$util$ClosureCleaner$$clean(ClosureCleaner.scala:305)
at org.apache.spark.util.ClosureCleaner$.clean(ClosureCleaner.scala:132)
at org.apache.spark.SparkContext.clean(SparkContext.scala:1891)
at org.apache.spark.streaming.dstream.DStream$$anonfun$map$1.apply(DStream.scala:528)
at org.apache.spark.streaming.dstream.DStream$$anonfun$map$1.apply(DStream.scala:528)
at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:148)
at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:109)
at org.apache.spark.SparkContext.withScope(SparkContext.scala:681)
at org.apache.spark.streaming.StreamingContext.withScope(StreamingContext.scala:258)
at org.apache.spark.streaming.dstream.DStream.map(DStream.scala:527)
at org.apache.spark.streaming.api.java.JavaDStreamLike$class.map(JavaDStreamLike.scala:157)
at org.apache.spark.streaming.api.java.AbstractJavaDStreamLike.map(JavaDStreamLike.scala:43)
at com.olacabs.spark.examples.WordCountProcessorKafkaImpl.process(WordCountProcessorKafkaImpl.java:45)
at com.olacabs.spark.examples.WordCountApp.main(WordCountApp.java:49)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:483)
at org.apache.spark.deploy.SparkSubmit$.org$apache$spark$deploy$SparkSubmit$$runMain(SparkSubmit.scala:664)
at org.apache.spark.deploy.SparkSubmit$.doRunMain$1(SparkSubmit.scala:169)
at org.apache.spark.deploy.SparkSubmit$.submit(SparkSubmit.scala:192)
at org.apache.spark.deploy.SparkSubmit$.main(SparkSubmit.scala:111)
at org.apache.spark.deploy.SparkSubmit.main(SparkSubmit.scala)
Caused by: java.io.NotSerializableException: Object of org.apache.spark.streaming.kafka.DirectKafkaInputDStream is being serialized possibly as a part of closure of an RDD operation. This is because the DStream object is being referred to from within the closure. Please rewrite the RDD operation inside this DStream to avoid this. This has been enforced to avoid bloating of Spark tasks with unnecessary objects.
Serialization stack:
- object not serializable (class: org.apache.spark.streaming.api.java.JavaStreamingContext, value: org.apache.spark.streaming.api.java.JavaStreamingContext@7add323c)
- field (class: com.olacabs.spark.examples.WordCountProcessorKafkaImpl, name: streamingContext, type: class org.apache.spark.streaming.api.java.JavaStreamingContext)
- object (class com.olacabs.spark.examples.WordCountProcessorKafkaImpl, com.olacabs.spark.examples.WordCountProcessorKafkaImpl@29a1505c)
- field (class: com.olacabs.spark.examples.WordCountProcessorKafkaImpl$1, name: this$0, type: class com.olacabs.spark.examples.WordCountProcessorKafkaImpl)
- object (class com.olacabs.spark.examples.WordCountProcessorKafkaImpl$1, com.olacabs.spark.examples.WordCountProcessorKafkaImpl$1@c6c82aa)
- field (class: org.apache.spark.api.java.JavaPairRDD$$anonfun$toScalaFunction$1, name: fun$1, type: interface org.apache.spark.api.java.function.Function)
- object (class org.apache.spark.api.java.JavaPairRDD$$anonfun$toScalaFunction$1, <function1>)
at org.apache.spark.serializer.SerializationDebugger$.improveException(SerializationDebugger.scala:40)
at org.apache.spark.serializer.JavaSerializationStream.writeObject(JavaSerializer.scala:47)
at org.apache.spark.serializer.JavaSerializerInstance.serialize(JavaSerializer.scala:81)
at org.apache.spark.util.ClosureCleaner$.ensureSerializable(ClosureCleaner.scala:312)
... 23 more
Can someone help me figure out why?
Here is the Code :
public interface EventProcessor extends Serializable {
void process();
}
public class WordCountProcessorKafkaImpl implements EventProcessor {
private static final Pattern SPACE = Pattern.compile(" ");
@Autowired
@Qualifier("streamingContext")
JavaStreamingContext streamingContext;
@Autowired
@Qualifier("inputDStream")
JavaPairInputDStream<String, String> inputDStream;
@Override
public void process() {
// Get the lines, split them into words, count the words and print
JavaDStream<String> lines = inputDStream.map(new Function<Tuple2<String, String>, String>() {
@Override
public String call(Tuple2<String, String> tuple2) {
return tuple2._2();
}
});
JavaDStream<String> words = lines.flatMap(new FlatMapFunction<String, String>() {
@Override
public Iterable<String> call(String x) {
return Lists.newArrayList(SPACE.split(x));
}
});
JavaPairDStream<String, Integer> wordCounts = words.mapToPair(
new PairFunction<String, String, Integer>() {
@Override
public Tuple2<String, Integer> call(String s) {
return new Tuple2<String, Integer>(s, 1);
}
}).reduceByKey(
new Function2<Integer, Integer, Integer>() {
@Override
public Integer call(Integer i1, Integer i2) {
return i1 + i2;
}
});
wordCounts.print();
// Start the computation
streamingContext.start();
streamingContext.awaitTermination();
}
}
Re: Serialization Exception
Posted by Tathagata Das <td...@databricks.com>.
I am guessing one of the two things might work.
1. Either define the pattern SPACE inside the process()
2. Mark streamingContext field and inputStream field as transient.
The problem is that the function like PairFunction needs to be serialized
for being sent to the tasks. And whole closure of the function is
serialized, and somehow that closure is capturing the whole
WordCountProcessorKafkaImpl
On Mon, Jun 29, 2015 at 5:14 AM, Spark Enthusiast <sp...@yahoo.in>
wrote:
> For prototyping purposes, I created a test program injecting dependancies
> using Spring.
>
> Nothing fancy. This is just a re-write of KafkaDirectWordCount. When I run
> this, I get the following exception:
>
> Exception in thread "main" org.apache.spark.SparkException: Task not
> serializable
> at
> org.apache.spark.util.ClosureCleaner$.ensureSerializable(ClosureCleaner.scala:315)
> at
> org.apache.spark.util.ClosureCleaner$.org$apache$spark$util$ClosureCleaner$$clean(ClosureCleaner.scala:305)
> at
> org.apache.spark.util.ClosureCleaner$.clean(ClosureCleaner.scala:132)
> at org.apache.spark.SparkContext.clean(SparkContext.scala:1891)
> at
> org.apache.spark.streaming.dstream.DStream$$anonfun$map$1.apply(DStream.scala:528)
> at
> org.apache.spark.streaming.dstream.DStream$$anonfun$map$1.apply(DStream.scala:528)
> at
> org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:148)
> at
> org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:109)
> at org.apache.spark.SparkContext.withScope(SparkContext.scala:681)
> at
> org.apache.spark.streaming.StreamingContext.withScope(StreamingContext.scala:258)
> at org.apache.spark.streaming.dstream.DStream.map(DStream.scala:527)
> at
> org.apache.spark.streaming.api.java.JavaDStreamLike$class.map(JavaDStreamLike.scala:157)
> at
> org.apache.spark.streaming.api.java.AbstractJavaDStreamLike.map(JavaDStreamLike.scala:43)
> at
> com.olacabs.spark.examples.WordCountProcessorKafkaImpl.process(WordCountProcessorKafkaImpl.java:45)
> at com.olacabs.spark.examples.WordCountApp.main(WordCountApp.java:49)
> at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
> at
> sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
> at
> sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
> at java.lang.reflect.Method.invoke(Method.java:483)
> at
> org.apache.spark.deploy.SparkSubmit$.org$apache$spark$deploy$SparkSubmit$$runMain(SparkSubmit.scala:664)
> at
> org.apache.spark.deploy.SparkSubmit$.doRunMain$1(SparkSubmit.scala:169)
> at org.apache.spark.deploy.SparkSubmit$.submit(SparkSubmit.scala:192)
> at org.apache.spark.deploy.SparkSubmit$.main(SparkSubmit.scala:111)
> at org.apache.spark.deploy.SparkSubmit.main(SparkSubmit.scala)
> Caused by: java.io.NotSerializableException: Object of
> org.apache.spark.streaming.kafka.DirectKafkaInputDStream is being
> serialized possibly as a part of closure of an RDD operation. This is
> because the DStream object is being referred to from within the closure.
> Please rewrite the RDD operation inside this DStream to avoid this. This
> has been enforced to avoid bloating of Spark tasks with unnecessary
> objects.
> Serialization stack:
> - object not serializable (class:
> org.apache.spark.streaming.api.java.JavaStreamingContext, value:
> org.apache.spark.streaming.api.java.JavaStreamingContext@7add323c)
> - field (class:
> com.olacabs.spark.examples.WordCountProcessorKafkaImpl, name:
> streamingContext, type: class
> org.apache.spark.streaming.api.java.JavaStreamingContext)
> - object (class
> com.olacabs.spark.examples.WordCountProcessorKafkaImpl,
> com.olacabs.spark.examples.WordCountProcessorKafkaImpl@29a1505c)
> - field (class:
> com.olacabs.spark.examples.WordCountProcessorKafkaImpl$1, name: this$0,
> type: class com.olacabs.spark.examples.WordCountProcessorKafkaImpl)
> - object (class
> com.olacabs.spark.examples.WordCountProcessorKafkaImpl$1,
> com.olacabs.spark.examples.WordCountProcessorKafkaImpl$1@c6c82aa)
> - field (class:
> org.apache.spark.api.java.JavaPairRDD$$anonfun$toScalaFunction$1, name:
> fun$1, type: interface org.apache.spark.api.java.function.Function)
> - object (class
> org.apache.spark.api.java.JavaPairRDD$$anonfun$toScalaFunction$1,
> <function1>)
> at
> org.apache.spark.serializer.SerializationDebugger$.improveException(SerializationDebugger.scala:40)
> at
> org.apache.spark.serializer.JavaSerializationStream.writeObject(JavaSerializer.scala:47)
> at
> org.apache.spark.serializer.JavaSerializerInstance.serialize(JavaSerializer.scala:81)
> at
> org.apache.spark.util.ClosureCleaner$.ensureSerializable(ClosureCleaner.scala:312)
> ... 23 more
>
>
>
> Can someone help me figure out why?
>
>
> Here is the Code :
>
> public interface EventProcessor extends Serializable {
> void process();
> }
>
>
> public class WordCountProcessorKafkaImpl implements EventProcessor {
>
> private static final Pattern SPACE = Pattern.compile(" ");
>
> @Autowired
> @Qualifier("streamingContext")
> JavaStreamingContext streamingContext;
>
> @Autowired
> @Qualifier("inputDStream")
> JavaPairInputDStream<String, String> inputDStream;
>
> @Override
> public void process() {
> // Get the lines, split them into words, count the words and print
> JavaDStream<String> lines = inputDStream.map(new Function<Tuple2<String, String>, String>() {
> @Override
> public String call(Tuple2<String, String> tuple2) {
> return tuple2._2();
> }
> });
> JavaDStream<String> words = lines.flatMap(new FlatMapFunction<String, String>() {
> @Override
> public Iterable<String> call(String x) {
> return Lists.newArrayList(SPACE.split(x));
> }
> });
> JavaPairDStream<String, Integer> wordCounts = words.mapToPair(
> new PairFunction<String, String, Integer>() {
> @Override
> public Tuple2<String, Integer> call(String s) {
> return new Tuple2<String, Integer>(s, 1);
> }
> }).reduceByKey(
> new Function2<Integer, Integer, Integer>() {
> @Override
> public Integer call(Integer i1, Integer i2) {
> return i1 + i2;
> }
> });
> wordCounts.print();
>
> // Start the computation
> streamingContext.start();
> streamingContext.awaitTermination();
>
> }
> }
>
>
>