Stream Processing in Java
A stream is a lambda (i.e. a function-object) that represents a sequence of values (objects or primitives). Each time it is called, it returns the next value in the sequence.
We can create a stream from a collection, array, or supplier (i.e. a parameterless lambda). For example, here is how to create a stream from a list:
List<Integer> numList = ...;
Stream<Integer> numStream = numList.stream();
Processing a stream means processing some or all of the values in the stream. (Processing a stream automatically calls the underlying lambda.) The result of processing a stream might be a resulting value (e.g., the sum), a side-effect (e.g., a printout), or another stream. For example:
Iterator<Integer> p = numStream.iterator();
while(p.hasNext()) { System.out.println(p.next()); }
Once a stream has been processed it
is closed and cannot be processed again.
Creating Streams
We can make streams from collections (sets and lists) and arrays. Here are a few others techniques:
Stream<String> silence = Stream.empty();
Stream<Double> noise = Stream.generate(Math::random);
Stream<Integer> nums = Stream.iterate(0, (n)->n + 1);
Stream Processors
A stream processor is a pipeline of filters and maps culminating in a reducer:
In the example above a stream is fed into a filter. A filter generates a new stream consisting of all values in the input stream that passed some test. (For example, filtering all
Next, the filtered stream is fed into a map. A map generates a new stream. Each value in the new stream is the result of transforming the corresponding value from the filtered stream by some transformer.
The output of the map is fed into another filter, which removes elements that fail to pass yet another test. These values are fed to another map where they are further transformed.
Finally the values are passed to a reducer. A reducer combines the values in a stream using some type of combiner.
Example: Sum of even squares
Problem: Compute the sum of squares of even numbers in a given list of integers:
Integer soes(List<Integer> nums) { ??? }
Solution: Let's implement soes as a stream processor:
The pipeline begins with a generator that converts a number list into a number stream. The filter removes odd numbers from the stream. The map transforms each value into its square. The reducer combines the values by adding them together. Instead of an integer, this last step produces an optional integer. An optional integer is an object that either contains the integer or a special null value if the integer couldn't be computed (for example if the number list was empty or contained something other than integers.) The last step extracts the integer if it's there.
Here's version 1.0 of the function:
public Integer soes(List<Integer> nums) {
Stream<Integer> numStream = nums.stream();
Stream<Integer> evens = numStream.filter((n)->n%2==0);
Stream<Integer> squares = evens.map((n)->n*n);
Optional<Integer> soesMaybe = squares.reduce((x,
y)->x+y);
if (soesMaybe.isPresent()) {
return
soesMaybe.get());
} else {
return
0;
}
}
Notice that a filter requires a Boolean-valued function to test values, map requires a transformer function to transform values, and reduce requires a combiner function to combine values.
The reducer returns an optional integer. This is a container that is either empty, because an error couldn't be computed for some reason, or the desired integer result.
Here's a slicker one-line version 2.0:
public Integer soes(List<Integer> nums) {
return
nums.stream().
filter(n -> n%2 == 0).
map(n
-> n*n).
reduce((n, m)-> n+m).
orElse(0);
}
Version 3.0 of soes uses old-fashioned list iteration:
public Integer soes(List<Integer> nums) {
Integer result = 0;
for(Integer n: nums) {
if (n % 2 == 0) {
result += n * n;
}
}
return result;
}
If nums contains 1000 integers, then this will require 4000 operations (%, ==, *, +=). If each operation takes a millisecond and if they are preformed sequentially, as they are in this version of soes, then it will take 4 seconds to compute the answer.
Here's version 4.0 of soes, which
uses a parallel stream:
public Integer soes(List<Integer> nums) {
return
nums.parallelStream().
filter(n -> n%2 == 0).
map(n
-> n*n).
reduce((n, m)-> n+m).
orElse(0);
}
This too will need to execute 4000 operations. But the Java VM can assign each stage of the pipeline to a different thread, so while the reducer is adding the first element of the stream to the resulting sum, the map is squaring the second element, and the filter is filtering the third element. Thus, the values can be processed three at-a-time, requiring 1.34 seconds.