Not Serializable exception when integrating SQL and Spark Streaming

Question

In Addition to Not Serializable exception when integrating Spark SQL and Spark Streaming

My source code

public static void main(String args[]) {
    SparkConf sparkConf = new SparkConf().setAppName("NumberCount");
    JavaSparkContext jc = new JavaSparkContext(sparkConf);
    JavaStreamingContext jssc = new JavaStreamingContext(jc, new Duration(2000));
    jssc.addStreamingListener(new WorkCountMonitor());
    int numThreads = Integer.parseInt(args[3]);
    Map<String,Integer> topicMap = new HashMap<String,Integer>();
    String[] topics = args[2].split(",");
    for (String topic : topics) {
        topicMap.put(topic, numThreads);
    }
    JavaPairReceiverInputDStream<String,String> data = KafkaUtils.createStream(jssc, args[0], args[1], topicMap);
    data.print();

    JavaDStream<Person> streamData = data.map(new Function<Tuple2<String, String>, Person>() {
            public Person call(Tuple2<String,String> v1) throws Exception {
                String[] stringArray = v1._2.split(",");
                Person Person = new Person();
                Person.setName(stringArray[0]);
                Person.setAge(stringArray[1]);
                return Person;
            }

        });


    final JavaSQLContext sqlContext = new JavaSQLContext(jc);
    streamData.foreachRDD(new Function<JavaRDD<Person>,Void>() {
        public Void call(JavaRDD<Person> rdd) {

            JavaSchemaRDD subscriberSchema = sqlContext.applySchema(rdd, Person.class);

            subscriberSchema.registerAsTable("people");
            System.out.println("all data");
            JavaSchemaRDD names = sqlContext.sql("SELECT name FROM people");
            System.out.println("afterwards");

            List<String> males = new ArrayList<String>();

            males = names.map(new Function<Row,String>() {
                public String call(Row row) {
                    return row.getString(0);
                }
            }).collect();
            System.out.println("before for");
            for (String name : males) {
                System.out.println(name);
            }
            return null;
        }
    });
    jssc.start();
    jssc.awaitTermination();
}

JavaSQLContext is also declared outside the ForeachRDD loop but i am still getting NonSerializableException

14/12/23 23:49:38 ERROR JobScheduler: Error running job streaming job 1419378578000 ms.1 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.rdd.RDD.map(RDD.scala:271) at org.apache.spark.api.java.JavaRDDLike$class.map(JavaRDDLike.scala:78) at org.apache.spark.sql.api.java.JavaSchemaRDD.map(JavaSchemaRDD.scala:42) at com.basic.spark.NumberCount$2.call(NumberCount.java:79) at com.basic.spark.NumberCount$2.call(NumberCount.java:67) at org.apache.spark.streaming.api.java.JavaDStreamLike$$anonfun$foreachRDD$1.apply(JavaDStreamLike.scala:274) at org.apache.spark.streaming.api.java.JavaDStreamLike$$anonfun$foreachRDD$1.apply(JavaDStreamLike.scala:274) at org.apache.spark.streaming.dstream.DStream$$anonfun$foreachRDD$1.apply(DStream.scala:529) at org.apache.spark.streaming.dstream.DStream$$anonfun$foreachRDD$1.apply(DStream.scala:529) at org.apache.spark.streaming.dstream.ForEachDStream$$anonfun$1.apply$mcV$sp(ForEachDStream.scala:42) at org.apache.spark.streaming.dstream.ForEachDStream$$anonfun$1.apply(ForEachDStream.scala:40) at org.apache.spark.streaming.dstream.ForEachDStream$$anonfun$1.apply(ForEachDStream.scala:40) at scala.util.Try$.apply(Try.scala:161) at org.apache.spark.streaming.scheduler.Job.run(Job.scala:32) at org.apache.spark.streaming.scheduler.JobScheduler$JobHandler.run(JobScheduler.scala:171) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615) at java.lang.Thread.run(Thread.java:724) Caused by: java.io.NotSerializableException: org.apache.spark.sql.api.java.JavaSQLContext at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1181) at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1541) at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1506) at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1429) at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1175) at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1541) at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1506) at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1429) at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1175) at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1541) at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1506) at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1429) at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1175) 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) ... 20 more

I appreciate if you have any suggestion.

Answer 1

Have you implemented Serializable interface in Person pojo class.Also can you try declaring topicMap as final

Answer 2

here is the working code

package com.basic.spark;

import java.io.Serializable;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;

import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.sql.api.java.JavaSQLContext;
import org.apache.spark.sql.api.java.JavaSchemaRDD;
import org.apache.spark.sql.api.java.Row;
import org.apache.spark.streaming.Duration;
import org.apache.spark.streaming.api.java.JavaDStream;
import org.apache.spark.streaming.api.java.JavaPairReceiverInputDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;
import org.apache.spark.streaming.kafka.KafkaUtils;

import scala.Tuple2;

public class NumberCount implements Serializable {

    transient SparkConf sparkConf = new SparkConf().setAppName("NumberCount");
    transient JavaSparkContext jc = new JavaSparkContext(sparkConf);
    transient JavaStreamingContext jssc_1 = new JavaStreamingContext(jc, new Duration(1000));
    transient JavaSQLContext sqlContext = new JavaSQLContext(jc);
    transient Producer producer = configureKafka();

    public static void main(String args[]) {
        (new NumberCount()).job_1(args);
    }

    public void job_1(String...args) {
        jssc_1.addStreamingListener(new WorkCountMonitor());
        int numThreads = Integer.parseInt(args[3]);
        Map<String,Integer> topicMap = new HashMap<String,Integer>();
        String[] topics = args[2].split(",");
        for (String topic : topics) {
            topicMap.put(topic, numThreads);
        }

        JavaPairReceiverInputDStream<String,String> data = KafkaUtils.createStream(jssc_1, args[0], args[1], topicMap);
        data.window(new Duration(10000), new Duration(2000));

        JavaDStream<String> streamData = data.map(new Function<Tuple2<String, String>, String>() {
            public String call(Tuple2<String,String> v1) {
                return v1._2;
            }
        });

        streamData.foreachRDD(new Function<JavaRDD<String>,Void>() {
            public Void call(JavaRDD<String> rdd) {

                if (rdd.count() < 1)
                    return null;

                try {
                    JavaSchemaRDD eventSchema = sqlContext.jsonRDD(rdd);
                    eventSchema.registerTempTable("event");
                    System.out.println("all data");
                    JavaSchemaRDD names = sqlContext.sql("SELECT deviceId, count(*) FROM event group by deviceId");
                    System.out.println("afterwards");

//                    List<Long> males = new ArrayList<Long>();
//
//                    males = names.map(new Function<Row,Long>() {
//                        public Long call(Row row) {
//                            return row.getLong(0);
//                        }
//                    }).collect();
//                    System.out.println("before for");
//                    ArrayList<KeyedMessage<String, String>> data = new ArrayList<KeyedMessage<String, String>>();
//                    for (Long name : males) {
//                        System.out.println("**************"+name);
//                        writeToKafka_1(data, String.valueOf(name));
//                    }
//                    producer.send(data);

                    List<String> deviceDetails = new ArrayList<String>();

                    deviceDetails = names.map(new Function<Row,String>() {
                        public String call(Row row) {
                            return row.getString(0) +":" + row.getLong(1);
                        }
                    }).collect();

                    System.out.println("before for");
                    ArrayList<KeyedMessage<String, String>> data = new ArrayList<KeyedMessage<String, String>>();
                    for (String name : deviceDetails) {
                        System.out.println("**************"+name);
                        writeToKafka_1(data, name);
                    }
                    producer.send(data);

                } catch (Exception e) {
                    System.out.println("#ERROR_1#   #" + rdd);
                    e.printStackTrace();
                }

                return null;
            }
        });
        jssc_1.start();
        jssc_1.awaitTermination();
    }

    public Producer<String, String> configureKafka() {
        Properties props = new Properties();
        props.put("metadata.broker.list", "xx.xx.xx.xx:9092");
        props.put("serializer.class", "kafka.serializer.StringEncoder");
        props.put("compression.codec", "2");
        props.put("request.required.acks", "0");
        props.put("producer.type", "sync");

        ProducerConfig config = new ProducerConfig(props);

        Producer<String, String> producer = new Producer<String, String>(config);

        return producer;
    }

    public void writeToKafka_1(ArrayList<KeyedMessage<String,String>> list, String msg) {
        list.add(new KeyedMessage<String,String>("my-replicated-topic-1", "", msg));
    }
}