Chapter 35. Streams

Distributed Streams work similarly to map reduce; however, with Distributed Streams there are zero to many intermediate operations followed by a single terminal operation that is sent to each node where work is performed. The following steps are used for this behavior:

When using distributed or replicated caches the keys and values must be marshallable; in addition, operations executed on Distributed Streams must also be marshallable, as these operations are sent to the other nodes in the cluster. This is most commonly accomplished by using a new class that is either Serializable or has an Externalizer registered; however, as the FunctionalInterface implements Serializable all lambdas are instantly serialized and thus no additional cast is required.

If a lambda function is in use this may be serialized by casting the parameter as an instance of Serializable. For instance, consider a cache that stores Book entries; the following would create a collection of Book instances that match a specific author:

List<Book> books = cache.keySet().stream()
    .filter((Predicate<? super Book>) e -> e.getAuthor().equals("authorname"))
    .collect(toList());

Additionally, not all produced Collectors are marshallable by default. JBoss Data Grid has included org.infinispan.stream.CacheCollectors as a convenient way to utilize any combination of Collectors that function properly when marshalling is required.

There are two different methods to parallelize streams:

By default, Distributed Streams enable parallel distribution; however, this may be further coupled with a parallel Stream, allowing concurrent operations executing across multiple nodes, with multiple threads on each node.

To mark a Stream as parallel it may either be obtained with parallelStream(), or it may be enabled after obtaining the Stream by invoking parallel(). The following example shows both methods:

// Obtain a parallel Stream initially
List<Book> books = cache.keySet().parallelStream()
    [...]

// Create the initial stream and then invoke parallel
List<Book> books = cache.keySet().stream()
    .parallel()
    [...]

A classic example of Map/Reduce is word count. Assuming we have a cache with String for keys and values, and we need to count the occurrence of all words in all sentences, this could be implemented using the following:

Map<String, Integer> wordCountMap = cache.entrySet().parallelStream()
    .map((Serializable & Function<Map.Entry<String, String>, String[]>) e -> e.getValue().split("\\s"))
    .flatMap((Function<String[], Stream<String>>) Arrays::stream)
    .collect(CacheCollectors.serializableCollector(() -> Collectors.groupingBy(Function.identity(), Collectors.counting())));

If we wanted to revise the example to find the most frequent word we would need to have all words available and counted locally first. The following snippet extends our previous example to perform this search:

String mostFrequentWord = cache.entrySet().parallelStream()
    .map((Serializable & Function<Map.Entry<String, String>, String[]>) e -> e.getValue().split("\\s"))
    .flatMap((Function<String[], Stream<String>>) Arrays::stream)
    .collect(CacheCollectors.serializableCollector(() -> Collectors.collectingAndThen(
        Collectors.groupingBy(Function.identity(), Collectors.counting()),
            wordCountMap -> {
                String mostFrequent = null;
                long maxCount = 0;
                for (Map.Entry<String, Long> e : wordCountMap.entrySet()) {
                    int count = e.getValue().intValue();
                    if (count > maxCount) {
                        maxCount = count;
                        mostFrequent = e.getKey();
                    }
                }
                return mostFrequent;
            })));

At present, this last step will be executed in a single thread. We can further optimize this operation by using a parallel stream locally to perform the final operation:

Map<String, Long> wordCount = cache.entrySet().parallelStream()
    .map((Function<Map.Entry<String, String>, String[]>) e -> e.getValue().split("\\s"))
    .flatMap((Function<String[], Stream<String>>) Arrays::stream)
    .collect(CacheCollectors.serializableCollector(() -> Collectors.groupingBy(Function.identity(), Collectors.counting())));
Optional<Map.Entry<String, Long>> mostFrequent = wordCount.entrySet().parallelStream()
    .reduce((e1, e2) -> e1.getValue() > e2.getValue() ? e1 : e2);