combiner argument on Stream.collect

[Pierrot Mongonnam]

Hi,

could somebody tell me what's the combiner argument on Stream.collect good for. Especially on the Oracle Java Tutorial an example:

1- an Averager Class wich should store the average of a range of ages:

class Averager implements IntConsumer
{
    private int total = 0;
    private int count = 0;
        
    public double average() {
        return count > 0 ? ((double) total)/count : 0;
    }
        
    public void accept(int i) { total += i; count++; }
    public void combine(Averager other) {
        total += other.total;
        count += other.count;
    }
}

2- then the code which computes the average of ages:

Averager averageCollect = roster.stream()
    .filter(p -> p.getGender() == Person.Sex.MALE)
    .map(Person::getAge)
    .collect(Averager::new, Averager::accept, Averager::combine);

So again my question: what does this combiner argument really do and why do I need it?

the explanation on the site is confusing to me:
combiner: The combiner function takes two result containers and merges their contents. In this example, it modifies an Averager result container by incrementing the count variable by the count member variable of the other Averager instance and adding to the total member variable the value of the other Averager instance's total member variable.

the explanation on Stream.collect is even worst!!!

Thanks for helping

[Marco Ehrentreich]

Hi Pierrot,

actually it's not as complicated as it may sound. Just a little bit strange if you're not familiar with functional programming concepts ;-)

The collect() method basically processes all the elements in the stream and allows you to define how they should be aggregated into a single result (container). In your example it takes an Averager object as an accumulator which contains the intermediate result (in this case the total sum of person's ages and the number of persons counted so far) and returns a new Averager object which represents the old result combined with the current element (= person) of the stream. That's why the third argument to the collect method is called "combiner" in the API documentation. Finally if you call average() on the resulting Averager object you get the average age of all counted persons.

I hope this explanation makes it a little bit clearer what is going on ;-)


Marco

[Pierrot Mongonnam]

Hi Marco,

many thanks for your quick answer. It's true that I'm not so experienced in functional programming in Java, but I think, I do understand the concept. You are explaining me the general idea of what is going on in the code. I've also understood that. But let's just focus on the Stream.collect method's arguments. for me the average process occurs already in the following fragment code of the Average class:

public double average() {
        return count > 0 ? ((double) total)/count : 0;
    }
        
public void accept(int i) { total += i; count++; }

So why do I still need the following combine method?

public void combine(Averager other) {
        total += other.total;
        count += other.count;
}

Sorry the silly question but as I said the code is somehow confusing for me. Also if you check the Documentation of the method Stream.collect(), the method has as we know, 3 arguments: a supplier, an accumulator and a combiner. So they say calling the method is like running the following code:

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

so again the silly question: where is this combiner coming into play?

Also in the Documentation,a weird explanation of the combiner:
combiner - an associative, non-interfering, stateless function for combining two values, which must be compatible with the accumulator function

In conclusion, since I don't understand the combiner I could not write my own result container from the scratch

Many thanks again for helping

[Marco Ehrentreich]

Aah, now I see what you mean ;-)

I must admit I don't know very much about implementation details of the stream API but it seems that with the current implementation in the Oracle JDK the combiner function is only used with parallel streams, i.e. if you'd call roster.parallelStream() instead of roster.stream(). You can easily verify this behavior by setting breakpoints for the methods in your Averager class and running the example in debug mode.

With parallel streams each partition of the whole stream accumulates its partial result in a separate instance of Averager. In this case the combiner function would be used to combine the partial result containers into a single Averager object.

But it seems there are no guarantees if and when the combiner argument is used so you shouldn't make the false assumption that it's only ever used for parallel streams.

Marco

[Pierrot Mongonnam]

Hi Marco,

what you are describing sounds very coherent. So I'll then try play with the code to get a better understanding. But by all means after your explanations I have a better feeling of what this curious method is doing. Many thanks for have helped!!! By the way I'm just curious are you from Germany? Many thanks again.

[Marco Ehrentreich]

You're welcome ;-) Glad that I could help you at least a little bit on that issue.

And yes, I'm from Germany. How did you know? I hope my English isn't THAT bad :-)

[Pierrot Mongonnam]

don't worry man. Dein Englisch ist hervorragend!!! No I just knew it from your name, because I'm also leaving in Germany . Danke deine Hilfe nochmals