Reduce vs Collect method on Parallel Streams vs Serial Streams

user_16073089 0
Viewed 107

I am failing to understand how reduce method work vs the collect method. Given
Set<String> set = new TreeSet<>(Set.of("b", "c", "d"));

Why does this code below produce firstResult= cdbcdb and secondResult= bcd

// Using reduce with StringBuilder
StringBuilder firstResult = set
        .parallelStream()
        .map(s -> new StringBuilder(s))
        // Requires Identity(StringBuilder), BiFunction, BinaryOperator
        .reduce(new StringBuilder(),
                (a, b) -> a.append(b),
                (a, b) -> a.append(b));

// Using collect with StringBuilder
StringBuilder secondResult = set
        .parallelStream()
        .map(s -> new StringBuilder(s))
        // collect requires Supplier, BiConsumer, BiConsumer
        .collect(StringBuilder::new,
                (a, b) -> a.append(b),
                (a, b) -> a.append(b));

while this code below produce firstResult=bdc and secondResult=''

// Using reduce with StringBuilder
StringBuilder firstResult = set
        .parallelStream()
        .map(s -> new StringBuilder(s))
        // Requires Identity(StringBuilder), BiFunction, BinaryOperator
        .reduce(new StringBuilder(),
                (a, b) -> b.append(a),
                (a, b) -> a.append(b));


// Using collect with StringBuilder
StringBuilder secondResult = set
        .parallelStream()
        .map(s -> new StringBuilder(s))
        // collect requires Supplier, BiConsumer, BiConsumer
        .collect(StringBuilder::new,
                (a, b) -> b.append(a),
                (a, b) -> a.append(b));

Thank you so much in advance!

java-stream java
1 Answers

To begin, the given program is modified to see what happen inside each scenario.

import java.util.Set;
import java.util.TreeSet;

public class ReduceVsCollect {
    public static void main(String[] args) {
        Set<String> set = new TreeSet<>(Set.of("b", "c", "d"));
        parallelReduceA_Append_B(set);
        parallelCollectA_Append_B(set);
        parallelReduceB_Append_A(set);
        parallelCollectB_Append_A(set);
    }

    private static void parallelReduceA_Append_B(Set<String> set) {
        System.out.println("-".repeat(80));
        System.out.println("Start parallelReduceA_Append_B");
        StringBuilder identity = new StringBuilder();
        System.out.println("identity hash: " + System.identityHashCode(identity));
        StringBuilder result = set
                .parallelStream()
                .map(s -> new StringBuilder(s))
                .reduce(identity,
                        (a, b) -> {
                            System.out.println(String.format("called accumulator, a hash:%s, a:%s, b:%s", System.identityHashCode(a), a, b));
                            return a.append(b);
                        },
                        (a, b) -> {
                            System.out.println(String.format("called combiner, a hash:%s, a:%s, b:%s", System.identityHashCode(a), a, b));
                            return a.append(b);
                        }
                );
        System.out.println("Result: " + result);
    }

    private static void parallelCollectA_Append_B(Set<String> set) {
        System.out.println("-".repeat(80));
        System.out.println("Start parallelCollectA_Append_B");
        StringBuilder result = set
                .parallelStream()
                .map(s -> new StringBuilder(s))
                .collect(StringBuilder::new,
                        (a, b) -> {
                            System.out.println(String.format("called accumulator, a hash:%s, a:%s, b:%s", System.identityHashCode(a), a, b));
                            a.append(b);
                        },
                        (a, b) -> {
                            System.out.println(String.format("called combiner, a hash:%s, a:%s, b:%s", System.identityHashCode(a), a, b));
                            a.append(b);
                        });
        System.out.println("Result: " + result);
    }

    private static void parallelReduceB_Append_A(Set<String> set) {
        System.out.println("-".repeat(80));
        System.out.println("Start parallelReduceB_Append_A");
        StringBuilder identity = new StringBuilder();
        System.out.println("identity hash: " + System.identityHashCode(identity));
        StringBuilder result = set
                .parallelStream()
                .map(s -> new StringBuilder(s))
                .reduce(identity,
                        (a, b) -> {
                            System.out.println(String.format("called accumulator, a hash:%s, a:%s, b:%s", System.identityHashCode(a), a, b));
                            return b.append(a);
                        },
                        (a, b) -> {
                            System.out.println(String.format("called combiner, a hash:%s, a:%s, b:%s", System.identityHashCode(a), a, b));
                            return b.append(a);
                        });
        System.out.println("Result: " + result);
    }

    private static void parallelCollectB_Append_A(Set<String> set) {
        System.out.println("-".repeat(80));
        System.out.println("Start parallelCollectB_Append_A");
        StringBuilder result = set
                .parallelStream()
                .map(s -> new StringBuilder(s))
                .collect(StringBuilder::new,
                        (a, b) -> {
                            System.out.println(String.format("called accumulator, a hash:%s, a:%s, b:%s", System.identityHashCode(a), a, b));
                            b.append(a);
                        },
                        (a, b) -> {
                            System.out.println(String.format("called combiner, a hash:%s, a:%s, b:%s", System.identityHashCode(a), a, b));
                            b.append(a);
                        });
        System.out.println("Result: " + result);
    }
}

Result

--------------------------------------------------------------------------------
Start parallelReduceA_Append_B
identity hash: 1642360923
called accumulator, a hash:1642360923, a:, b:c
called accumulator, a hash:1642360923, a:, b:b
called accumulator, a hash:1642360923, a:cb, b:d
called combiner, a hash:1642360923, a:cbd, b:cbd
Result: cbdcbd
--------------------------------------------------------------------------------
Start parallelCollectA_Append_B
called accumulator, a hash:1940447180, a:, b:c
called accumulator, a hash:572416449, a:, b:b
called accumulator, a hash:1940447180, a:c, b:d
called combiner, a hash:572416449, a:b, b:cd
Result: bcd
--------------------------------------------------------------------------------
Start parallelReduceB_Append_A
identity hash: 245565335
called accumulator, a hash:245565335, a:, b:c
called accumulator, a hash:245565335, a:, b:b
called accumulator, a hash:476402209, a:c, b:d
called combiner, a hash:1490180672, a:b, b:dc
Result: dcb
--------------------------------------------------------------------------------
Start parallelCollectB_Append_A
called accumulator, a hash:358699161, a:, b:c
called accumulator, a hash:1802696844, a:, b:b
called accumulator, a hash:358699161, a:, b:d
called combiner, a hash:1802696844, a:, b:
Result:

Explanation of different scenario:

1. Reduce with a.append(b) result in "cdbcdb" :

Repeated sequence is due to using mutable identity - new StringBuilder() in reduce(Refer to this answer). The value of identity keeps changing when accumulator is called. All intermediate result is also stored in identity, so when combiner is called, two sequence is shown.

2. Collect with a.append(b) result in "bcd" :

As Javadoc state,

Performs a mutable reduction operation on the elements of this stream. A mutable reduction is one in which the reduced value is a mutable result container, such as an ArrayList, and elements are incorporated by updating the state of the result rather than by replacing the result

Hence this is a valid scenario to use StringBuilder

3. Reduce with b.append(a) result in "bdc" :

This time identity is not modified due to swapping the parameter, and intermediate result is stored in the StringBuilder in .map(s -> new StringBuilder(s)), hence no repeated sequence is shown.

4. Collect with b.append(a) result in "" :

Refer to the psuedocode of collect,

     R result = supplier.get();
     for (T element : this stream)
         accumulator.accept(result, element);
     return result;

In the accumulator, we called b.append(a) which means result(a) is never modified, so it keeps its initial value as new StringBuilder().

user_79287210
Related