Styjun

I'm doing Spark Streaming to parse some kafka messages in real time. Before I parse the message, I read some file from local and construct two variables GridMatrix GM and LinkMatcher LM which are useful to parsing. Here is the code that gives me java.lang.StackOverflowError when I submit it using spark-submit xxx.jar:

public class Stream implements Serializable 
 GridMatrix GM = GridMatrixConstructor.init_Grid_Matrix(0.001);
 LinkMatcher LM = new LinkMatcher();

 public void parse_rdd_record(String[] fields) 
 try 
 System.out.println(InetAddress.getLocalHost().getHostName() + "---->" + Thread.currentThread());
 
 catch (Exception e) 
 e.printStackTrace();
 
 System.out.println(LM.GF.toString());
 System.out.println(GM.topleft_x);
 

 public void Streaming_process() throws Exception 
 SparkConf conf = new SparkConf()
 .setAppName("SparkStreaming")
 .setMaster("local[*]");
 conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer");
 conf.registerKryoClasses(new Class<?>[]
 Class.forName("Streaming.Stream")
 );


 JavaSparkContext sc = new JavaSparkContext(conf);
 sc.setLogLevel("WARN");
 JavaStreamingContext ssc = new JavaStreamingContext(sc, new Duration(2000));
 Map<String, Object> kafkaParams = new HashMap<>();
 kafkaParams.put("bootstrap.servers", "xxx.xx.xx.xx:20103,xxx.xx.xx.xx:20104,xxx.xx.xx.xx:20105");
 kafkaParams.put("key.deserializer", StringDeserializer.class);
 kafkaParams.put("value.deserializer", StringDeserializer.class);
 kafkaParams.put("group.id", "use_a_separate_group_id_for_each_stream");
 kafkaParams.put("auto.offset.reset", "latest");
 kafkaParams.put("enable.auto.commit", false);

 Collection<String> topics = Arrays.asList("nc_topic_gis_test");
 JavaInputDStream<ConsumerRecord<String, String>> GPS_DStream =
 KafkaUtils.createDirectStream(
 ssc,
 LocationStrategies.PreferConsistent(),
 ConsumerStrategies.<String, String>Subscribe(topics, kafkaParams)
 );

 JavaPairDStream<String, String> GPS_DStream_Pair = GPS_DStream.mapToPair(
 (PairFunction<ConsumerRecord<String, String>, String, String>) record ->
 new Tuple2<>("GPSValue", record.value()));

 GPS_DStream_Pair.foreachRDD(PairRDD -> PairRDD.foreach(rdd -> 
 String[] fields = rdd._2.split(",");
 this.parse_rdd_record(fields);
 ));

 ssc.start();
 ssc.awaitTermination();
 

 public static void main(String[] args) throws Exception 
 new Stream().Streaming_process();
 

It gives me the following error:

Exception in thread "streaming-job-executor-0" java.lang.StackOverflowError
 at java.io.Bits.putDouble(Bits.java:121)
 at java.io.ObjectStreamClass$FieldReflector.getPrimFieldValues(ObjectStreamClass.java:2168)
 at java.io.ObjectStreamClass.getPrimFieldValues(ObjectStreamClass.java:1389)
 at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1533)
 at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509)
 at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432)
 at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178)
 at java.io.ObjectOutputStream.writeArray(ObjectOutputStream.java:1378)
 at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1174)
 at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548)
 at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509)
 at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432)
 at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178)
 at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548)
 at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509)
 at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432)
 at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178)
 at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548)
 at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509)
 at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432)
 at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178)
 at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548)
 at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509)
 at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432)
 at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178)
 at java.io.ObjectOutputStream.writeObject(ObjectOutputStream.java:348)
 at java.util.HashMap.internalWriteEntries(HashMap.java:1790)
 at java.util.HashMap.writeObject(HashMap.java:1363)
 at sun.reflect.GeneratedMethodAccessor37.invoke(Unknown Source)
 at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
 at java.lang.reflect.Method.invoke(Method.java:498)
 at java.io.ObjectStreamClass.invokeWriteObject(ObjectStreamClass.java:1140)
 at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1496)
 at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432)
 at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178)
 at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548)
 at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509)
 at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432)
 at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178)
 at java.io.ObjectOutputStream.writeObject(ObjectOutputStream.java:348)
 at java.util.HashMap.internalWriteEntries(HashMap.java:1790)
 at java.util.HashMap.writeObject(HashMap.java:1363)
 at sun.reflect.GeneratedMethodAccessor37.invoke(Unknown Source)
 at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)

However, if I change the GM and LM to static variable, it runs well. Change line 2 and line 3 to:

private static final GridMatrix GM = GridMatrixConstructor.init_Grid_Matrix(0.001);
private static final LinkMatcher LM = new LinkMatcher();

Could someone tell me why it won't work with none static variables?

The difference between static and non static version is that when non static it sends them to all workers as the Stream closure, when static not by default, except it using by one of streaming lambda, which is not a case.

While sending to the workers it trying to serialize an object. And it fails according to the provided stack trace. The reason most probably because those structures has cyclic declaration on themselfs inside and can not be properly serializable.