Distributed streams implement standard Java interface, so there are plenty of sources on the topic 'How to use Java streams?'. It's true that there's little info in docs (@wburns, have I missed that?), the best source ATM is this blogpost: Infinispan: Distributed Streams - most notable difference is that Java lambdas are not serializable without additional effort, and standard Collectors aren't serializable either.

To solve your problem, you need to make DoubleSummaryStatistics serializable/externalizable, I think that the stacktrace is quite self-explanatory. It may be not so clear how to solve that, though; please consult Marshalling chapter in Infinispan User Guide to find out how an Externalizer or AdvancedExternalizer can be registered.

Unfortunately the documentation piece I was working on and due to personal reasons I wasn't able to complete that before I had to leave.  I am back now though and this should be in before 8.2.0 Final is out.

Also you already answered your concern about the "confusing" stack trace when you stated

Marius Grama wrote:

 

(WeatherDataStatistics is serializable, DoubleSummaryStatistics on the other hand is not )

Unfortunately DoubleSummaryStatistics is not Serializable by default.  I looked at this a bit when I was first implementing distributed streams and didn't see a good way of doing this, using reflection in an Externalizer seems like the only realistic way.  Instead I was thinking of possibly adding another method to CacheCollectors to replace the Collectors methods that use the SummaryStatistics and also the summaryStatistics methods from the DoubleStream, IntStream, LongStream implementations.  What do you think?

The key part here to realize that when using a distributed stream is that we will attempt to distribute as much as feasible across the cluster.  So the confusing part may be figuring out which part is returned and finally done as the final collect phase.  This is done in between your collectingAndThen method.  The downstream collector is distributed and the results from that are sent to the owning node where it does a final reduction using the finisher.

The problem (both streams and map/reduce) is you need an intermediate structure that would allow for reduction in an associative way to be applied that is also Serializable.  Unfortunately the JRE by default doesn't some a structure.

The simplest way I can see for your case is to use a double[] that is 2 wide (one element containing the count and another the running count).  You can then use the reducing method as the downstream collector of collectingAndThen.  Then for the finisher Function in the andThen clause you can calculate the average from those numbers and have all the data available you wanted.

Marius Grama wrote:

 

How can the problem of calculating averages in the cache can be solved via streams?

I answered yours above, but I also want to clarify something else.  Looking at your code, you are doing something different than the Map/Reduce tutorial.  In your case you are returning sum and average for each country, where as Map/Reduce is returning only the average.  Returning multiple values like that is a bit more cumbersome when using Map/Reduce (it would also need this intermediate container).

If you want a simple streams example that does exactly what the map reduce does in the embedded tutorial, this is all that is needed:

Map<String, Double> resultMap = cache.entrySet().stream()
     .collect(CacheCollectors.serializableCollector(() -> groupingBy(Map.Entry::getKey,
              averagingDouble(e -> e.getValue().temperature))));

Marius Grama wrote:

 

It seems to me (I'm a newbie with infinispan) that the MapReduce utilities still make sense in infinispan because it is much more straightforward from a developer perspective how to get to calculate the average. Doesn't it seem not justified the deprecation of the mapreduce utilities from infinispan?

There are quite a few reasons for this actually.

1. While the map/reduce API is very similar to Hadoop and others, the new streams is using the new Java APIs directly.  This will gain more traction and usage as the years go on and the latter will most likely dwindle.
2. The new streams in our testing is faster for every use case we have ran so far.  Larger data sets especially it starts performing many orders of magnitude faster.
3. The Map/Reduce implementation is not rehash aware.  Meaning that if a node leaves or joins while the computation is running you will almost always get the wrong value.  Streams supports this and even allows disabling if needed.

I also have since created [ISPN-6191] Provide Externalizers for (Double|Long|Int)SummaryStatistics - JBoss Issue Tracker.  I convinced myself this go around that this is the easiest way.  This should be fixed in 8.2.0, so you will be able to use any of the collector methods that return back summary statistics.

Great, glad to hear you got it working.

Marius Grama wrote:

 

 

I was trying several possibilities on how to get the average and this is how I got to calculating the double summary statistics, when there was a much simpler solution at hand (like the one you've mentioned).

 

On the other hand concerning the XSummaryStatistics serialization trick, I'm not sure that this is really needed. See whether it will not introduce new problems down the road with it just for giving syntactic sugar usage for the stream manipulation.

I still think that having a exhaustive and mature set of code samples (with the downside of having to maintain them) would be very helpful for the users of infinispan.

Yes for the averages the small code snippet will work for just averages.  But if you want to return more than that, say also the count or sum, you need the intermediate container.  Because of that I have gone the way of providing Serialization for the various SummaryStatistics classes in the previous JIRA I mentioned.

I agree on the examples, the tutorial will be updated at some point in the near future to use streams instead.

Thanks for your input!