引发异常的火花sortby

Question

我正在尝试按键对JavaPairRDD进行排序。

 JavaPairRDD<String, HashMap<String, Object>> countAndSum = grupBydate
                                        .reduceByKey(new Function2<HashMap<String, Object>, HashMap<String, Object>, HashMap<String, Object>>() {
                                            @Override
                                            public HashMap<String, Object> call(
                                                    HashMap<String, Object> v1,
                                                    HashMap<String, Object> v2)
                                                    throws Exception {
                                                long count = Long.parseLong(v1.get(
                                                        SparkToolConstant.COUNT)
                                                        .toString())
                                                        + Long.parseLong(v2
                                                                .get(SparkToolConstant.COUNT)
                                                                .toString());
                                                Double sum = Double.parseDouble(v1.get(
                                                        SparkToolConstant.VALUE)
                                                        .toString())
                                                        + Double.parseDouble(v2
                                                                .get(SparkToolConstant.VALUE)
                                                                .toString());
                                                HashMap<String, Object> sumMap = new HashMap<String, Object>();
                                                sumMap.put(SparkToolConstant.COUNT,
                                                        count);
                                                sumMap.put(SparkToolConstant.VALUE, sum);
                                                return sumMap;
                                            }
                                        });


System.out.println("count before sorting : "
                                        + countAndSum.count());



   /**
    sort by date 

    */
                                JavaPairRDD<String, HashMap<String, Object>> sortByDate = countAndSum
                                        .sortByKey(new Comparator<String>() {
                                            @Override
                                            public int compare(String dateStr1,
                                                    String dateStr2) {
                                                DateUtil dateUtil = new DateUtil();
                                                Date date1 = dateUtil.stringToDate(
                                                        dateStr1, dateFormat);
                                                Date date2 = dateUtil.stringToDate(
                                                        dateStr2, dateFormat);
                                                if (date2 == null && date1 == null) {
                                                    return 0;
                                                } else if (date2 != null
                                                        && date1 != null) {
                                                    return date1.compareTo(date2);
                                                } else if (date2 == null) {
                                                    return 1;
                                                } else {
                                                    return -1;
                                                }
                                            }
                                        });

                        System.out.println("count after sorting : "
                                + sortByDate.count());

块引号

Answer 1

Spark将首先序列化您在reducebykey和sorbykey中传递的函数，并将它们传递给执行器。因此，您应该保证您的函数在那里是可序列化的

代码中的sparktoolconstant&dateutil似乎是导致此错误的原因。