Spark kryo\u序列化程序和广播java。io。IOException：java。lang.UnsupportedOperationException

Question

当我尝试访问广播变量时，我会收到这个异常：

2006年3月17日03:04:23警告TaskSetManager：在第3.0阶段（TID 10，192.168.56.5，executor 1）中丢失任务0.0：java。io。IOException：java。lang.UnsupportedOperationException位于组织。阿帕奇。火花util。Utils美元。组织上的tryOrIOException（Utils.scala:1276）。阿帕奇。火花广播TorrentBroadcast。org上的readBroadcastBlock（TorrentBroadcast.scala:206）。阿帕奇。火花广播TorrentBroadcast_价值$lzycompute（TorrentBroadcast.scala:66）在org上。阿帕奇。火花广播TorrentBroadcast_org上的值（TorrentBroadcast.scala:66）。阿帕奇。火花广播TorrentBroadcast。getValue（TorrentBroadcast.scala:96），售价为2美元。在org上调用（GowallaTask.java:214）。阿帕奇。火花api。Java语言JavaRDDLike$$anonfun$foreach$1。在org上应用（JavaRDDLike.scala:351）。阿帕奇。火花api。Java语言JavaRDDLike$$anonfun$foreach$1。在scala上应用（JavaRDDLike.scala:351）。收集迭代器$类。org上的foreach（迭代器.scala:893）。阿帕奇。火花中断迭代器。foreach（interruptableiterator.scala:28）位于org。阿帕奇。火花rdd。RDD$$anonfun$foreach$1$$anonfun$apply$28。在组织中应用（RDD.scala:917）。阿帕奇。火花rdd。RDD$$anonfun$foreach$1$$anonfun$apply$28。在组织中应用（RDD.scala:917）。阿帕奇。火花SparkContext$$anonfun$runJob$5。在org上应用（SparkContext.scala:1944）。阿帕奇。火花SparkContext$$anonfun$runJob$5。在org上应用（SparkContext.scala:1944）。阿帕奇。火花调度程序。结果任务。在组织上运行任务（ResultTask.scala:87）。阿帕奇。火花调度程序。任务在组织上运行（Task.scala:99）。阿帕奇。火花执行人。执行者$TaskRunner。在java上运行（Executor.scala:282）。util。同时发生的线程池执行器。java上的runWorker（ThreadPoolExecutor.java:1142）。util。同时发生的ThreadPoolExecutor$工作者。在java上运行（ThreadPoolExecutor.java:617）。lang.Thread。运行（Thread.java:745）

原因：java。java上的lang.UnsupportedOperationException。util。抽象地图。将（AbstractMap.java:209）放在com上。深奥的软件。克鲁约。序列化程序。MapSerializer。在com上读取（MapSerializer.java:162）。深奥的软件。克鲁约。序列化程序。MapSerializer。在com上读取（MapSerializer.java:39）。深奥的软件。克鲁约。克鲁约。org上的readClassAndObject（Kryo.java:790）。阿帕奇。火花序列化程序。KryoDeserializationStream。org上的readObject（KryoSerializer.scala:244）。阿帕奇。火花广播TorrentBroadcast$$anonfun 10美元。在org上申请（TorrentBroadcast.scala:286）。阿帕奇。火花util。Utils美元。tryWithSafeFinally（Utils.scala:1303）位于org。阿帕奇。火花广播TorrentBroadcast美元。在org上解锁对象（TorrentBroadcast.scala:287）。阿帕奇。火花广播TorrentBroadcast$$anonfun$readBroadcastBlock$1。申请（TorrentBroadcast.scala:221），网址：org。阿帕奇。火花util。Utils美元。tryOrIOException（Utils.scala:1269）。。。19个以上

我使用KryoSerializer时收到异常

    conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer");
    conf.set("spark.kryoserializer.buffer.mb", "24");

这是我的密码。

JavaPairRDD<Object, Iterable<GowallaDataLocation>> line_RDD_2 = sc
            .textFile("/home/piero/gowalla_location.txt", 2).map(new GowallaMapperDataLocation())
            .groupBy(new Function<GowallaDataLocation, Object>() {

                /**
                 * 
                 */
                private static final long serialVersionUID = -6773509902594100325L;

                @Override
                public Object call(GowallaDataLocation v1) throws Exception {
                    DateFormat dateFormat = new SimpleDateFormat("yyyyMMdd");

                    return dateFormat.format(v1.getDATE());
                }
            }).persist(StorageLevel.MEMORY_AND_DISK_SER());



Broadcast<Map<Object, Iterable<GowallaDataLocation>>> broadcastVar_2 = sc.broadcast(line_RDD_2.collectAsMap());
    //System.out.println(broadcastVar_2.getValue().size());

    JavaRDD<Object> keys = line_RDD_2.keys().persist(StorageLevel.MEMORY_ONLY_SER());
    line_RDD_2.unpersist();

    keys.foreach(new VoidFunction<Object>() {

        /**
         * 
         */
        private static final long serialVersionUID = -8148877518271969523L;

        @Override
        public void call(Object t) throws Exception {
            // TODO Auto-generated method stub
            //System.out.println("KEY:" + t + " ");
            Iterable<GowallaDataLocation> dr = broadcastVar_2.getValue().get(t);

        }

    });

Answer 1

我怀疑这是因为您直接广播了line_RDD_2.collectAsMap（）：这意味着广播的类型是Map，kryo不知道正确的实现，将使用AbstractMap进行内部工作

就像我这样做：

Map<String, String> a = new HashMap<String, String>();
a.put("a", "b");
Set<String> c = a.keySet();
c.add("e");

我将获得不支持的AbstractCollection操作，很容易解决：

Map<String, String> a = new HashMap<String, String>();
a.put("a", "b");
Set<String> c = new TreeSet<String>();
c.addAll(a.keySet());
c.add("e");

如果我的猜测是对的，您可能可以这样解决：

Map<Object, Iterable<GowallaDataLocation>> a = new HashMap<>();
a.putAll(line_RDD_2.collectAsMap());
Broadcast<Map<Object, Iterable<GowallaDataLocation>>> broadcastVar_2 = sc.broadcast(a);

让我知道这是否有效