Whenever a collect() is called in a distributed environment, you have to wrap up in a serializable supplier method call to make it serializable, e.g.

Map<Integer, Long> totalPerHour = cache.values().stream()
   .collect(
       CacheCollectors.serializableCollector(() -> Collectors.groupingBy(
           e -> getHourOfDay(e.departureTs),
           Collectors.counting()
       )));

This is a bit clunky. We should overload collect to take a SerializableSupplier in CacheStream.

It would also be nice to have an overloaded version that takes Supplier in case you want to provide your own serialization for it.