第35章 ストリーム

分散ストリームは、map reduce と同様に動作します。しかし、分散ストリームでは、ゼロから多数の中間操作があり、その後にワークが実行される各ノードへ送信される単一の終端操作が続きます。この動作には以下の手順が使用されます。

分散キャッシュまたはレプリケートされたキャッシュを使用する場合、キーと値がマーシャル可能である必要があります。また、分散ストリームで実行される操作はクラスターの他のノードに送信されるため、これらの操作もマーシャル可能である必要があります。これらをマーシャル可能にするには、一般的に Serializable である新しいクラスを使用するか、Externalizer が登録済みの新しいクラスを使用します。しかし、FunctionalInterface は Serializable を実装し、すべてのラムダは即座にシリアライズされるため、追加のキャストは必要ありません。

ラムダ関数が使用されている場合、パラメーターを Serializable のインスタンスとしてキャストするとシリアライズできます。たとえば、Book エントリーを格納するキャッシュの場合、以下は特定の著者と一致する Book インスタンスのコレクションを作成します。

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

さらに、デフォルトでは作成されたすべての Collectors がマーシャル可能であるとは限りません。JBoss Data Grid には、マーシャリングが必要なときに適切に機能する Collectors の組み合わせを容易に使用できる org.infinispan.stream.CacheCollectors が含まれています。

ストリームを並列処理するメソッドは 2 つあります。

デフォルトでは、分散ストリームは並列分散を有効にしますが、さらに並列 Stream に結合できます。これにより、各ノードで複数のスレッドを使用して、複数のノード全体で同時操作を実行できます。

Stream を並列としてマーク付けするには、parallelStream() で取得するか、parallel() を呼び出して Stream の取得後に有効にします。以下の例は両方の方法を示しています。

// 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()
    [...]

典型的な Map/Reduce の例が単語数のカウントです。キーと値に対して String を持つキャッシュがあり、文すべての単語数を数える必要がある場合、以下を使用して実装できます。

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())));

例を変更して最も頻繁に使用される単語を見つける場合、すべての単語が必要になり、最初にローカルで数える必要があります。以下のスニペットは前述の例を拡張し、この検索を実行します。

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;
            })));

この時点で、最後のステップが単一のスレッドで実行されます。ローカルで並列ストリームを使用して最終の操作を実行すると、この操作をさらに最適化できます。

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);