欢迎您访问程序员文章站本站旨在为大家提供分享程序员计算机编程知识!
您现在的位置是: 首页

spark 程序 org.apache.spark.SparkException: Task not serializable

程序员文章站 2022-07-15 12:55:24
...

1.看代码,报上面这个错,原因可参考 https://www.cnblogs.com/zwCHAN/p/4305156.html

public class StreamDemo implements Runnable {

    Map<String, Map<String, String>> mapping;

    public void init() {
        mapping = DaoUtil.getColumnRelation();
    }

    public void exet() {
        Map<String, Object> kafkaParams = new HashMap<>();
        kafkaParams.put("bootstrap.servers", "centos-1:9092,centos-2:9092,centos-3:9092");
        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);
        Map<String, Map<String, String>> mapping = DaoUtil.getColumnRelation();;
        SparkConf conf = new SparkConf().setMaster("local[2]").setAppName("kafka-syn");
        JavaStreamingContext context = new JavaStreamingContext(conf, Durations.seconds(5));

        Collection<String> topics = Arrays.asList("real-table-data");

        JavaInputDStream<ConsumerRecord<String, String>> stream =
                KafkaUtils.createDirectStream(
                        context,
                        LocationStrategies.PreferConsistent(),
                        ConsumerStrategies.<String, String>Subscribe(topics, kafkaParams)
                );
        stream.mapToPair(record -> new Tuple2<>(record.key(), record.value())).foreachRDD(new VoidFunction2<JavaPairRDD<String, String>, Time>() {
            @Override
            public void call(JavaPairRDD<String, String> str, Time time) throws Exception {
                str.foreach(new VoidFunction<Tuple2<String, String>>() {
                    @Override
                    public void call(Tuple2<String, String> stringStringTuple2) throws Exception {
                        System.out.println(stringStringTuple2._2);
                    }
                });
            }
        });

        context.start();
        try {
            context.awaitTermination();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    @Override
    public void run() {
        init();
        exet();
    }

    public static void main(String[] args) {
        Thread thread = new Thread(new StreamDemo());
        thread.start();
    }
}

2.修改实现,实现序列化接口,问题解决

public class StreamDemo implements Runnable,Serializable {
相关标签: spark 序列化