As we mentioned earlier in this chapter, the reduce operation applies a summary operation to the elements of a stream to generate a single summary result. This single result can be of the same type as the elements of the stream of an other type. A simple example of the reduce operation is calculating the sum of a stream of numbers.

The stream API provides the reduce() method to implement reduction operations. This method has the following three different versions:

The reduce() method has a limitation. As we mentioned before, it must return a single value. You shouldn't use the reduce() method to generate a collection or a complex object. The first problem is performance. As the documentation of the stream API specifies, the accumulator function returns a new value every time it processes an element. If your accumulator function works with collections, every time it processes an element it creates a new collection, which is very inefficient. Another problem is that, if you work with parallel streams, all the threads will share the identity value. If this value is a mutable object, for example, a collection, all the threads will be working over the same collection. This does not comply with the philosophy of the reduce() operation. In addition, the combiner() method will always receive two identical collections (all the threads are working over only one collection), that also doesn't comply the philosophy of the reduce() operation.

If you want to make a reduction that generates a collection or a complex object, you have the following two options:

In this example, we will use the reduce() method to obtain information about the inverted index and the forEach() method to reduce the index to the list of relevant documents for a query.

In our first approach, we will use all the documents associated with a word. The steps of this implementation of our search process are:

We have implemented this version in the basicSearch()method of the ConcurrentSearch class. This is the source code of the method:

        public static void basicSearch(String query[]) throws IOException {

        Path path = Paths.get("index", "invertedIndex.txt");
        HashSet<String> set = new HashSet<>(Arrays.asList(query));
        QueryResult results = new QueryResult(new ConcurrentHashMap<>());

        try (Stream<String> invertedIndex = Files.lines(path)) {

            invertedIndex.parallel() .filter(line -> set.contains(Utils.getWord(line))) .flatMap(ConcurrentSearch::basicMapper) .forEach(results::append);

            results .getAsList() .stream() .sorted() .limit(100) .forEach(System.out::println);

            System.out.println("Basic Search Ok");
        }

    }

We receive an array of string objects with the words of the query. First, we transform that array in a set. Then, we use a try-with-resources stream with the lines of the invertedIndex.txt file, which is the file that contains the inverted index. We use a try-with-resources so we don't have to worry about opening or closing the file. The aggregate operations of the stream will generate a QueryResult object with the relevant documents. We use the following methods to obtain that list:

Once we have created the QueryResult object, we create other streams to obtain the final list of results using the following methods:

Let's describe the auxiliary classes and methods used in the example.

    public static Stream<Token> basicMapper(String input) {
        ConcurrentLinkedDeque<Token> list = new ConcurrentLinkedDeque();
        String word = Utils.getWord(input);
        Arrays .stream(input.split(","))
          .skip(1) .parallel() .forEach(token -> list.add(new Token(word, token)));

        return list.stream();
    }

public class Token {

    private final String word;
    private final double tfxidf;
    private final String file;

    public Token(String word, String token) {
        this.word=word;
        String[] parts=token.split(":");
        this.file=parts[0];
        this.tfxidf=Double.parseDouble(parts[1]);
    }

    @Override
    public String toString() {
        return word+":"+file+":"+tfxidf;
    }

public class QueryResult {

    private Map<String, Document> results;

    public QueryResult(Map<String, Document> results) {
        this.results=results;
    }

    public void append(Token token) {
        results.computeIfAbsent(token.getFile(), s -> new Document(s)).addTfxidf(token.getTfxidf());
    }

    public List<Document> getAsList() {
        return new ArrayList<>(results.values());
    }

The first approach creates a new Token object per word and file. We have noted that common words, for example, the, take a lot of documents associated and a lot of them have low values of tfxidf. We have changed our mapper method to take into account only 100 files per word, so the number of Token objects generated will be smaller.

We have implemented this version in the reducedSearch() method of the ConcurrentSearch class. This method is very similar to the basicSearch() method. It only changes the stream operations that generate the QueryResult object, as follows:

        invertedIndex.parallel() .filter(line -> set.contains(Utils.getWord(line))) .flatMap(ConcurrentSearch::limitedMapper) .forEach(results::append);

Now, we use the limitedMapper() method as function in the flatMap() method.

    public static Stream<Token> limitedMapper(String input) {
        ConcurrentLinkedDeque<Token> list = new ConcurrentLinkedDeque();
        String word = Utils.getWord(input);

        Arrays.stream(input.split(",")) .skip(1) .limit(100) .parallel() .forEach(token -> {
            list.add(new Token(word, token));
          });

        return list.stream();
    }

While working with a search tool using a web search engine (for example, Google), when you make a search, it returns you the results of your search (the 10 most important) and for every result the title of the document and a fragment of the document where the words you have searched for appears.

Our third approach to the search tool is based on the second approach but by adding a third stream to generate an HTML file with the results of the search. For every result, we will show the title of the document and three lines where one of the words introduced in the query appears. To implement this, you need access to the files that appears in the inverted index. We have stored them in a folder named docs.

This third approach is implemented in the htmlSearch()method of the ConcurrentSearch class. The first part of the method to construct the QueryResult object with the 100 results is equal to the reducedSearch() method, as follows:

    public static void htmlSearch(String query[], String fileName) throws IOException {
        Path path = Paths.get("index", "invertedIndex.txt");
        HashSet<String> set = new HashSet<>(Arrays.asList(query));
        QueryResult results = new QueryResult(new ConcurrentHashMap<>());

        try (Stream<String> invertedIndex = Files.lines(path)) {

            invertedIndex.parallel() .filter(line -> set.contains(Utils.getWord(line))) .flatMap(ConcurrentSearch::limitedMapper) .forEach(results::append);

Then, we create the file to write the output and the HTML headers in it:

                         path = Paths.get("output", fileName + "_results.html");
            try (BufferedWriter fileWriter = Files.newBufferedWriter(path, StandardOpenOption.CREATE)) {

                fileWriter.write("<HTML>");
                fileWriter.write("<HEAD>");
                fileWriter.write("<TITLE>");
                fileWriter.write("Search Results with Streams");
                fileWriter.write("</TITLE>");
                fileWriter.write("</HEAD>");
                fileWriter.write("<BODY>");
                fileWriter.newLine();

Then, we include the stream that generates the results in the HTML file:

                            results.getAsList()
                    .stream()
                    .sorted()
                    .limit(100)
                    .map(new ContentMapper(query)).forEach(l -> {
                        try {
                            fileWriter.write(l);
                            fileWriter.newLine();
                        } catch (IOException e) {
                            e.printStackTrace();
                        }
                });

                fileWriter.write("</BODY>");
                fileWriter.write("</HTML>");

            }

We have used the following methods:

Let's see the details of the ContentMapper class.

public class ContentMapper implements Function<Document, String> {
    private String query[];

    public ContentMapper(String query[]) {
        this.query = query;
    }

    public String apply(Document d) {
        String result = "";
        
        try (Stream<String> content = Files.lines(Paths.get("docs",d.getDocumentName()))) {
            result = "<h2>" + d.getDocumentName() + ": "
                    + content.findFirst().get()
                    + ": " + d.getTfxidf() + "</h2>";
        } catch (IOException e) {
            e.printStackTrace();
            throw new UncheckedIOException(e);
        }

                try (Stream<String> content = Files.lines(Paths.get ("docs",d.getDocumentName()))) {
            result += content
                    .filter(l -> Arrays.stream(query).anyMatch (l.toLowerCase()::contains))
                    .limit(3)
                    .map(l -> "<p>"+l+"</p>")
                    .reduce("",String::concat);
            return result;
        } catch (IOException e) {
            e.printStackTrace();
            throw new UncheckedIOException(e);
        }
    }

The three previous solutions have a problem when they are executed in parallel. As we mentioned earlier, parallel streams are executed using the common Fork/Join pool provided by the Java concurrency API. In Chapter 6, Optimizing Divide and Conquer Solutions – The Fork/Join Framework, you learned that you shouldn't use I/O operations as read or write data in a file inside the tasks. This is because when a thread has blocked reading or writing data from or to a file, the framework doesn't use the work-stealing algorithm. As we use a file as the source of our streams, we are penalizing our concurrent solution.

One solution to this problem is to read the data to a data structure and then create our streams from that data structure. Obviously, the execution time of this approach will be smaller when we compare it with the other approaches, but we want to compare the serial and concurrent versions to see (as we expect) that the concurrent version gives us better performance than the serial version. The bad part of this approach is that you need to have your data structure in memory, so you will need a big amount of memory.

This fourth approach is implemented in the preloadSearch() method of the ConcurrentSearch class. This method receives the query as an Array of String and an object of the ConcurrentInvertedIndex class (we will see the details of this class later) with the data of the inverted index as parameters. This is the source code of this version:

        public static void preloadSearch(String[] query, ConcurrentInvertedIndex invertedIndex) {

        HashSet<String> set = new HashSet<>(Arrays.asList(query));
        QueryResult results = new QueryResult(new ConcurrentHashMap<>());

        invertedIndex.getIndex()
            .parallelStream()
            .filter(token -> set.contains(token.getWord()))
            .forEach(results::append);

        results
            .getAsList()
            .stream()
            .sorted()
            .limit(100)
            .forEach(document -> System.out.println(document));

        System.out.println("Preload Search Ok.");
    }

The ConcurrentInvertedIndex class has List<Token> to store all the Token objects read from the file. It has two methods, get() and set() for this list of elements.

As in other approaches, we use two streams: the first one to get a ConcurrentLinkedDeque of Result objects with the whole list of results and the second one to write the results in the console. The second one doesn't change over other versions, but the first one changes. We use the following methods in this stream:

public class ConcurrentFileLoader {

    public ConcurrentInvertedIndex load(Path path) throws IOException {
        ConcurrentInvertedIndex invertedIndex = new ConcurrentInvertedIndex();
        ConcurrentLinkedDeque<Token> results=new ConcurrentLinkedDeque<>();

        try (Stream<String> fileStream = Files.lines(path)) {
            fileStream
            .parallel()
            .flatMap(ConcurrentSearch::limitedMapper)
            .forEach(results::add);
        }
        
        invertedIndex.setIndex(new ArrayList<>(results));
        return invertedIndex;
    }
}

To go further with this example, we're going to test another concurrent version. As we mentioned in the introduction of this chapter, parallel streams use the common Fork/Join pool introduced in Java 8. However, we can use a trick to use our own pool. If we execute our method as a task of the Fork/Join pool, all the operations of the stream will be executed in the same Fork/Join pool. To test this functionality, we have added the executorSearch() method to the ConcurrentSearch class. This method receives the query as an array of String objects as a parameter, the InvertedIndex object, and a ForkJoinPool object. This is the source code of this method:

    public static void executorSearch(String[] query, ConcurrentInvertedIndex invertedIndex, ForkJoinPool pool) {
        HashSet<String> set = new HashSet<>(Arrays.asList(query));
        QueryResult results = new QueryResult(new ConcurrentHashMap<>());

        pool.submit(() -> {
            invertedIndex.getIndex()
                .parallelStream()
                .filter(token -> set.contains(token.getWord()))
                .forEach(results::append);

            results
                .getAsList()
                .stream()
                .sorted()
                .limit(100)
                .forEach(document -> System.out.println(document));
        }).join();

        System.out.println("Executor Search Ok.");

    }

We execute the content of the method, with its two streams, as a task in the Fork/Join pool using the submit() method, and wait for its finalization using the join() method.

We have implemented some methods to get information about the inverted index using the reduce() method in the ConcurrentData class.

The first method calculates the number of words in a file. As we mentioned earlier in this chapter, the inverted index stores the files in which a word appears. If we want to know the words that appear in a file, we have to process all the inverted index. We have implemented two versions of this method. The first one is implemented in getWordsInFile1(). It receives the name of the file and the InvertedIndex object as parameters, as follows:

    public static void getWordsInFile1(String fileName, ConcurrentInvertedIndex index) {
        long value = index
                .getIndex()
                .parallelStream()
                .filter(token -> fileName.equals(token.getFile()))
                .count();
        System.out.println("Words in File "+fileName+": "+value);
    }

In this case, we get the list of Token objects using the getIndex() method and create a parallel stream using the parallelStream() method. Then, we filter the tokens associated with the file using the filter() method, and finally, we count the number of words associated with that file using the count() method.

We have implemented another version of this method using the reduce() method instead of the count() method. It's the getWordsInFile2() method:

    public static void getWordsInFile2(String fileName, ConcurrentInvertedIndex index) {

        long value = index
                .getIndex()
                .parallelStream()
                .filter(token -> fileName.equals(token.getFile()))
                .mapToLong(token -> 1)
                .reduce(0, Long::sum);
        System.out.println("Words in File "+fileName+": "+value);
    }

The start of the sequence of operations is the same as the previous one. When we have obtained the stream of Token objects with the words of the file, we use the mapToInt() method to convert that stream into a stream of 1 and then the reduce() method to sum all the 1 numbers.

We have implemented the getAverageTfxidf() method that calculates the average tfxidf value of the words of a file in the collection. We have used here the reduce() method to show how it works. You can use other methods here with better performance:

    public static void getAverageTfxidf(String fileName, ConcurrentInvertedIndex index) {

        long wordCounter = index
                .getIndex()
                .parallelStream()
                .filter(token -> fileName.equals(token.getFile()))
                .mapToLong(token -> 1)
                .reduce(0, Long::sum);

        double tfxidf = index
                .getIndex()
                .parallelStream()
                .filter(token -> fileName.equals(token.getFile()))
                .reduce(0d, (n,t) -> n+t.getTfxidf(), (n1,n2) -> n1+n2);

        System.out.println("Words in File "+fileName+": "+(tfxidf/wordCounter));
    }

We use two streams. The first one calculates the number of words in a file and has the same source code as the getWordsInFile2() method. The second one calculates the total tfxidf value of all the words in the file. We use the same methods to get the stream of Token objects with the words in the file and then we use the reduce method to sum the tfxidf value of all the words. We pass the following three parameters to the reduce() method:

We have also used the reduce() method to calculate the maximum and minimum tfxidf values of the inverted index in the maxTfxidf() and minTfxidf() methods:

    public static void maxTfxidf(ConcurrentInvertedIndex index) {
        Token token = index
                .getIndex()
                .parallelStream()
                .reduce(new Token("", "xxx:0"), (t1, t2) -> {
                    if (t1.getTfxidf()>t2.getTfxidf()) {
                        return t1;
                    } else {
                        return t2;
                    }
                });
        System.out.println(token.toString());
    }

The method receives the ConcurrentInvertedIndex as a parameter. We use the getIndex() to obtain the list of Token objects. Then, we use the parallelStream() method to create a parallel stream over the list and the reduce() method to obtain the Token with the biggest tfxidf. In this case, we use the reduce() method with two parameters: an identity value and an accumulator function. The identity value is a Token object. We don't care about the word and the file name, but we initialize its tfxidf attribute with the value 0. Then, the accumulator function receives two Token objects as parameters. We compare the tfxidf attribute of both objects and return the one with greater value.

The minTfxidf() method is very similar, as follows:

    public static void minTfxidf(ConcurrentInvertedIndex index) {
        Token token = index
                .getIndex()
                .parallelStream()
                .reduce(new Token("", "xxx:1000000"), (t1, t2) -> {
                    if (t1.getTfxidf()<t2.getTfxidf()) {
                        return t1;
                    } else {
                        return t2;
                    }
                });
        System.out.println(token.toString());
    }

The main difference is that in this case, the identity value is initialized with a very high value for the tfxidf attribute.

To test all the methods explained in the previous sections, we have implemented the ConcurrentMain class that implements the main() method to launch our tests. In these tests, we have used the following three queries:

We have tested the three queries with the three versions of our search process measuring the execution time of each test. All the tests have a code similar to this:

public class ConcurrentMain {

    public static void main(String[] args) {

        String query1[]={"james","bond"};
        String query2[]={"gone","with","the","wind"};
        String query3[]={"rocky"};
        
            Date start, end;

        bufferResults.append("Version 1, query 1, concurrent
");
        start = new Date();
        ConcurrentSearch.basicSearch(query1);
        end = new Date();
        bufferResults.append("Execution Time: "
                + (end.getTime() - start.getTime()) + "
");

To load the inverted index from a file to an InvertedIndex object, you can use the following code:

        ConcurrentInvertedIndex invertedIndex = new ConcurrentInvertedIndex();
        ConcurrentFileLoader loader = new ConcurrentFileLoader();
        invertedIndex = loader.load(Paths.get("index","invertedIndex.txt"));

To create the Executor to use in the executorSearch() method, you can use the following code:

        ForkJoinPool pool = new ForkJoinPool();

We have implemented a serial version of this example with the SerialSearch, SerialData, SerialInvertendIndex, SerialFileLoader, and SerialMain classes. To implement that version, we have made the following changes:

Let's compare the solutions of the serial and concurrent versions of all the methods we have implemented. We have executed them using the JMH framework (http://openjdk.java.net/projects/code-tools/jmh/), which allows you to implement micro benchmarks in Java. Using a framework for benchmarking is a better solution that simply measures time using methods such as currentTimeMillis() or nanoTime(). We have executed them 10 times in a computer with a four-core processor so a concurrent algorithm can become theoretically four times faster than a serial one. Take into account that we have implemented a special class to execute the JMH tests. You can find these classes in the com.javferna.packtpub.mastering.irsystem.benchmark package of the source code.

For the first query, with the words james and bond, these are the execution times obtained in milliseconds:

For the second query, with the words gone, with, the, and wind, these are the execution times obtained in milliseconds:

For the third query, with the words rocky, these are the execution times obtained in milliseconds:

Finally, these are the average execution times in milliseconds for the methods that return information about the inverted index:

We can draw the following conclusions:

We can compare the parallel and sequential streams for the three queries in this end using the speed-up:

Finally, in our third approach, we generate an HTML web page with the results of the queries. These are the first results with the query james bond:

For the query gone with the wind, these are the first results:

Finally, these are the first results for the query rocky: