Producing a sorted wordcount with Spark I'm currently learning how to use Apache Spark. In order to do so, I implemented a simple wordcount (not really original, I know). There already exists an example on the documentation providing the needed code. However, I tried to go one step further and to produce the result as a sorted list (by frequence) by using the spark modules. 

A use case would be sortedWordcount("I fish a fish") -&gt;  [(fish, 2), (I,1), (a,1))]

Here is how I proceed:

public List&lt;String&gt; sortedWordcount() {
    SparkConf conf = new SparkConf().setAppName("wordcount").setMaster("local");
    JavaSparkContext sc = new JavaSparkContext(conf);

    JavaRDD&lt;String&gt; lines = sc.textFile("myFile.txt");

    // classical wordcount, nothing new there
    JavaRDD&lt;String&gt; words = lines.flatMap(new FlatMapFunction&lt;String, String&gt;() {
                public Iterable&lt;String&gt; call(String s) {
                    return Arrays.asList(s.split(" "));
                    }
                });
    JavaPairRDD&lt;String, Integer&gt; pairs = words.mapToPair(new PairFunction&lt;String, String, Integer&gt;() {
                public Tuple2&lt;String, Integer&gt; call(String s) {
                    return new Tuple2&lt;String, Integer&gt;(s, 1);
                }
            });
        JavaPairRDD&lt;String, Integer&gt; counts = pairs.reduceByKey(new Function2&lt;Integer, Integer, Integer&gt;() {
                public Integer call(Integer a, Integer b) {
                    return a + b;
                }
            });

    // to enable sorting by value (count) and not key -&gt; value-to-key conversion pattern
    JavaPairRDD&lt;Tuple2&lt;Integer, String&gt;, Integer&gt; countInKey = counts.mapToPair(a -&gt; new Tuple2(new Tuple2&lt;Integer, String&gt;(a._2, a._1), null)); // setting value to null, since it won't be used anymore

    List&lt;Tuple2&lt;Tuple2&lt;Integer, String&gt;, Integer&gt;&gt; wordSortedByCount = countInKey.sortByKey(new TupleComparator(), false).collect();

    List&lt;String&gt; result = new ArrayList&lt;&gt;();
    IntStream.range(0, wantedSize).forEach(i -&gt; result.add(wordSortedByCount.get(i)._1._2));
    return result;
}

private class TupleComparator implements Comparator&lt;Tuple2&lt;Integer, String&gt;&gt;, Serializable {
    @Override
    public int compare(Tuple2&lt;Integer, String&gt; tuple1, Tuple2&lt;Integer, String&gt; tuple2) {
        return tuple1._1 &lt; tuple2._1 ? 0 : 1;
    }
}


To sum up on this code work: 


it stores the text as a list of word
it changes this list of word into a list of tuple, each tuple being (word, 1) with word as key
it combine tuples with the same key and counts them, tuples are now (word, freqWord)
it changes the way tuples are displayed, they now become ((freqWord, word), null)
tuples are sorted by key (see the Comparator implementation)


My main concern is about performances, but I'll gladly welcome any critiques on any aspects of this code. Is there any way to achieve the same goal in less / more efficient steps?
I'll review this code with keeping in mind that you want to leave Apache Spark intact, as you intend to use it in the future for purposes it is intended to do.
I strongly believe that your current code can be written in pure Java 8, though that is not the point of the question.

Use lambdas wherever you can!

This will greatly enhance readability of your code.

As an example:

JavaRDD&lt;String&gt; words = lines.flatMap(new FlatMapFunction&lt;String, String&gt;() {
            public Iterable&lt;String&gt; call(String s) {
                return Arrays.asList(s.split(" "));
                }
            });


could be rewritten as the following (I checked and FlatMapFunction is indeed an interface that qualifies to be a functional interface):

JavaRDD&lt;String&gt; words = lines.flatMap(str -&gt; Arrays.asList(s.split(" ")));


Similarly it holds that:

JavaPairRDD&lt;String, Integer&gt; pairs = words.mapToPair(new PairFunction&lt;String, String, Integer&gt;() {
            public Tuple2&lt;String, Integer&gt; call(String s) {
                return new Tuple2&lt;String, Integer&gt;(s, 1);
            }
        });


can be rewritten as:

JavaPairRDD&lt;String, Integer&gt; pairs = words.mapToPair(str -&gt; new Tuple2&lt;&gt;(s, 1));


If type inference is not strong enough, then you need to write: new Tuple2&lt;String, Integer&gt;(s, 1).

Lastly it also holds for:

    JavaPairRDD&lt;String, Integer&gt; counts = pairs.reduceByKey(new Function2&lt;Integer, Integer, Integer&gt;() {
            public Integer call(Integer a, Integer b) {
                return a + b;
            }
        });


can be written as:

JavaPairRDD&lt;String, Integer&gt; counts = pairs.reduceByKey((a, b) -&gt; a + b);


or even better:

JavaPairRDD&lt;String, Integer&gt; counts = pairs.reduceByKey(Integer::sum);


Use the new way to write comparators

You can much much easier write comparators in Java 8, it works via key extraction, you can turn this code:

private class TupleComparator implements Comparator&lt;Tuple2&lt;Integer, String&gt;&gt;, Serializable {
    @Override
    public int compare(Tuple2&lt;Integer, String&gt; tuple1, Tuple2&lt;Integer, String&gt; tuple2) {
        return tuple1._1 &lt; tuple2._1 ? 0 : 1;
    }
}


into this code:

Comparator&lt;Tuple2&lt;Integer, String&gt;&gt; tupleComparator = Comparator.comparing(tuple2 -&gt; tuple2._1);


This will automatically and automagically create a working comparator that works by comparator the _1 field of tuple2, which is inferred to be Tuple2&lt;Integer, String&gt;.

Now this line:

List&lt;Tuple2&lt;Tuple2&lt;Integer, String&gt;, Integer&gt;&gt; wordSortedByCount = countInKey.sortByKey(new TupleComparator(), false).collect();


can become:

List&lt;Tuple2&lt;Tuple2&lt;Integer, String&gt;, Integer&gt;&gt; wordSortedByCount =
countInKey.sortByKey(Comparator.comparing(tuple2 -&gt; tuple2._1, false).collect();


Normally one would use method references in case of Comparator.comparing, if Tuple2 had a get1() method, then you would write it as Comparator.comparing(Tuple2::get1).

Upon further inspecting your code I see that you possibly intentionally have made your comparator serializable, if you need that for Apache Spark, then you can do that as well, though the solution does get a fair bit uglier then:

List&lt;Tuple2&lt;Tuple2&lt;Integer, String&gt;, Integer&gt;&gt; wordSortedByCount = 
countInKey.sortByKey((Comparator&lt;Tuple2&lt;Integer, String&gt;&gt; &amp; Serializable)Comparator.comparing(tuple2 -&gt; tuple2._1, false).collect();


Via this mechanism it is possible to make a type implement a wrapper interface without writing boilerplate. It is more common in the form of (Runnable &amp; Serializable)someRunnable than in this form though.

Final words

Lastly your code formatting is in need for some improvements with respect to indenting the anonymous classes, but it may be an oversight and the rest of the code looks great.

When I have some time I'd be eager to show you a pure Java 8 solution as I strongly believe that Apache Spark is making a mess of your code currently. I do understand however that you need to operate with Apache Spark at some point and for that you may be forced to use Apache Spark's classes.

My method using Java 8

As addendum I'll show how I would identify your problem in question and show you how I would do it.

Input: An input file, consisting of words.
Output: A list of the words sorted by frequency in which they occur.

Map&lt;String, Long&gt; occurenceMap = Files.readAllLines(Paths.get("myFile.txt"))
        .stream()
        .flatMap(line -&gt; Arrays.stream(line.split(" ")))
        .collect(Collectors.groupingBy(i -&gt; i, Collectors.counting()));
List&lt;String&gt; sortedWords = occurenceMap.entrySet()
        .stream()
        .sorted(Comparator.comparing((Map.Entry&lt;String, Long&gt; entry) -&gt; entry.getValue()).reversed())
        .map(Map.Entry::getKey)
        .collect(Collectors.toList());


This will do the following steps:


Read all lines into a List&lt;String&gt; (care with large files!)
Turn it into a Stream&lt;String&gt;.
Turn that into a Stream&lt;String&gt; by flat mapping every String to a Stream&lt;String&gt; splitting on the blanks.
Collect all elements into a Map&lt;String, Long&gt; grouping by the identity (i -&gt; i) and using as downstream Collectors.counting() such that the map-value will be its count.
Get a Set&lt;Map.Entry&lt;String, Long&gt;&gt; from the map.
Turn it into a Stream&lt;Map.Entry&lt;String, Long&gt;&gt;.
Sort by the reverse order of the value of the entry.
Map the results to a Stream&lt;String&gt;, you lose the frequency information here.
Collect the stream into a List&lt;String&gt;.


Beware that the line .sorted(Comparator.comparing((Map.Entry&lt;String, Long&gt; entry) -&gt; entry.getValue()).reversed()) should really be .sorted(Comparator.comparing(Map.Entry::getValue).reversed(), but type inference is having issues with that and for some reason it will not compile.

I hope the Java 8 way can give you interesting insights.