集成 SQL 和 Spark Streaming 时出现不可序列化异常

Question

除了Not Serializable exception when integrating Spark SQL and Spark Streaming

我的源代码

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 也在 ForeachRDD 循环之外声明，但我仍然收到 NonSerializableException

14/12/23 23:49:38 错误 JobScheduler：运行作业流作业时出错 1419378578000 ms.1 org.apache.spark.SparkException：任务不可序列化 在 org.apache.spark.util.ClosureCleaner$.ensureSerializable(ClosureCleaner.scala:166) 在 org.apache.spark.util.ClosureCleaner$.clean(ClosureCleaner.scala:158) 在 org.apache.spark.SparkContext.clean(SparkContext.scala:1435) 在 org.apache.spark.rdd.RDD.map(RDD.scala:27​​1) 在 org.apache.spark.api.java.JavaRDDLike$class.map(JavaRDDLike.scala:78) 在 org.apache.spark.sql.api.java.JavaSchemaRDD.map(JavaSchemaRDD.scala:42) 在 com.basic.spark.NumberCount$2.call(NumberCount.java:79) 在 com.basic.spark.NumberCount$2.call(NumberCount.java:67) 在 org.apache.spark.streaming.api.java.JavaDStreamLike$$anonfun$foreachRDD$1.apply(JavaDStreamLike.scala:27​​4) 在 org.apache.spark.streaming.api.java.JavaDStreamLike$$anonfun$foreachRDD$1.apply(JavaDStreamLike.scala:27​​4) 在 org.apache.spark.streaming.dstream.DStream$$anonfun$foreachRDD$1.apply(DStream.scala:529) 在 org.apache.spark.streaming.dstream.DStream$$anonfun$foreachRDD$1.apply(DStream.scala:529) 在 org.apache.spark.streaming.dstream.ForEachDStream$$anonfun$1.apply$mcV$sp(ForEachDStream.scala:42) 在 org.apache.spark.streaming.dstream.ForEachDStream$$anonfun$1.apply(ForEachDStream.scala:40) 在 org.apache.spark.streaming.dstream.ForEachDStream$$anonfun$1.apply(ForEachDStream.scala:40) 在 scala.util.Try$.apply(Try.scala:161) 在 org.apache.spark.streaming.scheduler.Job.run(Job.scala:32) 在 org.apache.spark.streaming.scheduler.JobScheduler$JobHandler.run(JobScheduler.scala:171) 在 java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145) 在 java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615) 在 java.lang.Thread.run(Thread.java:724) 引起：java.io.NotSerializableException：org.apache.spark.sql.api.java.JavaSQLContext 在 java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1181) 在 java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1541) 在 java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1506) 在 java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1429) 在 java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1175) 在 java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1541) 在 java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1506) 在 java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1429) 在 java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1175) 在 java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1541) 在 java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1506) 在 java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1429) 在 java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1175) 在 java.io.ObjectOutputStream.writeObject(ObjectOutputStream.java:347) 在 org.apache.spark.serializer.JavaSerializationStream.writeObject(JavaSerializer.scala:42) 在 org.apache.spark.serializer.JavaSerializerInstance.serialize(JavaSerializer.scala:73) 在 org.apache.spark.util.ClosureCleaner$.ensureSerializable(ClosureCleaner.scala:164) ... 20 更多

如果您有任何建议，我将不胜感激。

Answer 1

你是否在 Person pojo 类中实现了 Serializable 接口。你也可以尝试将 topicMap 声明为 final

Answer 2

这是工作代码

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));
    }
}