12

I'm trying to sort an RDD by value, and if multiple values are equal then I need to these values by key lexicographically.

code :

JavaPairRDD <String,Long> rddToSort = rddMovieReviewReducedByKey.mapToPair(new PairFunction < Tuple2 < String, MovieReview > , String, Long > () {

    @Override
    public Tuple2 < String, Long > call(Tuple2 < String, MovieReview > t) throws Exception {
        return new Tuple2 < String, Long > (t._1, t._2.count);
    }
});

What I have done so far is this, using takeOrdered and providing a CustomComperator, but since takeOrdered can't handle a large amount of data, when running the code it keeps exiting (it eats a lot of memory that the OS can't handle) :

List < Tuple2 < String, Long >> rddSorted = rddMovieReviewReducedByKey.mapToPair(new PairFunction < Tuple2 < String, MovieReview > , String, Long > () {

    @Override
    public Tuple2 < String, Long > call(Tuple2 < String, MovieReview > t) throws Exception {
        return new Tuple2 < String, Long > (t._1, t._2.count);
    }
}).takeOrdered(newTopMovies, MapLongValueComparator.VALUE_COMP);

Comperator:

    static class MapLongValueComparator implements Comparator < Tuple2 < String, Long >> , Serializable {
        private static final long serialVersionUID = 1L;

        private static final MapLongValueComparator VALUE_COMP = new MapLongValueComparator();

        @Override
        public int compare(Tuple2 < String, Long > o1, Tuple2 < String, Long > o2) {
            if (o1._2.compareTo(o2._2) == 0) {
                return o1._1.compareTo(o2._1);
            }
            return -o1._2.compareTo(o2._2);
        }
}

ERROR:

16/06/30 21:09:23 INFO scheduler.DAGScheduler: Job 18 failed: takeOrdered at MovieAnalyzer.java:708, took 418.149182 s

How would you sort this RDD? How would you take the TopKMovies considering value, and in case of equality keys lexicographically.

Thanks.

6
  • Can you provide the stack trace (if there is any?). Because you mentioned that it might be the memory problem, but the error message does not allows to see what exactly happened.
    – Serhiy
    Jul 1, 2016 at 7:28
  • @Serhiy I guess it is a memory problem since takeOrdered operation takes a long time, because it is handling large amount of data in Distributed mode, I got Exit code :137 & Exit code : 1 as well. approaching the sort in other way will definitely solve the problem.
    – Jay
    Jul 1, 2016 at 8:09
  • Have you tried re-partitioning the data? When you map to pair, you can re-partition right afterwards.
    – Serhiy
    Jul 1, 2016 at 9:41
  • 1
    Partitioning will split your data into multiple parts, so the machine, with configured restrictions can rather work on parts and not on entire data which at some point will stop fitting into machine memory (do not forget Spark does in-memory computations). Right after mapping the values to pair you can call partitionBy method. You will need to implemented your partitioner through, which in very simple case could simply partition by the first letter of the String (which I guess is a movie name?). You might need to experiment further with partitioning in case it will still go OOM.
    – Serhiy
    Jul 1, 2016 at 9:57
  • 1
    It may solve the timeout problem that takeOrdered does on a large amount of data, but that won't sort the data correctly!
    – Jay
    Jul 1, 2016 at 12:50

2 Answers 2

3

Solved the problem using sortByKey with a comparator & partitions, after maping the <String, Long> PairRDD to < Tuple2<String,Long> , Long> PairRDD

JavaPairRDD <Tuple2<String,Long>, Long> sortedRdd = rddMovieReviewReducedByKey.mapToPair(new PairFunction < Tuple2 < String, MovieReview > , Tuple2<String,Long>, Long > () {

    @Override
    public Tuple2 < Tuple2<String,Long>, Long > call(Tuple2 < String, MovieReview > t) throws Exception {
        return new Tuple2 < Tuple2<String,Long>, Long > (new Tuple2<String,Long>(t._1,t._2.count), t._2.count);
    }
}).sortByKey(new TupleMapLongComparator(), true, 100);


JavaPairRDD <String,Long> sortedRddToPairs = sortedRdd.mapToPair(new PairFunction<Tuple2<Tuple2<String,Long>,Long>, String, Long>() {

    @Override
    public Tuple2<String, Long> call(
            Tuple2<Tuple2<String, Long>, Long> t) throws Exception {
        return new Tuple2 < String, Long > (t._1._1, t._1._2);
    }

});

Comparator:

private class TupleMapLongComparator implements Comparator<Tuple2<String,Long>>, Serializable {
    @Override
    public int compare(Tuple2<String,Long> tuple1, Tuple2<String,Long> tuple2) {

        if (tuple1._2.compareTo(tuple2._2) == 0) {
            return tuple1._1.compareTo(tuple2._1);
        }
        return -tuple1._2.compareTo(tuple2._2);
    }
}
1

did you try secondary sorting in Spark ?

Spark Secondary Sort

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge that you have read and understand our privacy policy and code of conduct.

Not the answer you're looking for? Browse other questions tagged or ask your own question.