Streams and Collectors

A stream is a sequence of elements that a chain of operations processes, one element at a time, from a source such as a collection. This lesson builds on Lambdas and Functional Interfaces and the collections in Lists, Deques, and Iteration and HashMap Internals. After it, a reader can write and read stream pipelines and predict when each operation runs.

Idea or API	Release
`Stream`, `Collectors`, `IntStream`, `LongStream`, `DoubleStream`	Java 8
`Stream.iterate` with a predicate, `takeWhile`, `dropWhile`	Java 9
`Collectors.teeing`	Java 12
`Stream.toList`, `Stream.mapMulti`	Java 16

Example 1: a pipeline is lazy

A pipeline has three parts: a source, zero or more intermediate operations (which return a new stream), and one terminal operation (which produces a result or a side effect). Intermediate operations do nothing until the terminal operation starts.

Java

import java.util.List;
import java.util.stream.Stream;

public class LazyPipeline {
    public static void main(String[] args) {
        Stream<String> pipeline = Stream.of("a", "bb", "ccc")
                .filter(s -> { System.out.println("filter " + s); return s.length() > 1; })
                .map(s -> { System.out.println("map " + s); return s.toUpperCase(); });
        System.out.println("pipeline built, nothing ran yet");

        List<String> result = pipeline.toList();
        System.out.println(result);
    }
}

Output

pipeline built, nothing ran yet
filter a
filter bb
map bb
filter ccc
map ccc
[BB, CCC]

Building the pipeline prints nothing. When toList runs, each element goes through filter and then map before the next element starts. The stream does not run all of filter first and then all of map. This element-by-element order is called vertical processing, and it is why a pipeline needs no intermediate lists.

Example 2: short-circuiting and stateful operations

A short-circuiting operation can finish without reading the whole source: findFirst, anyMatch, limit, and takeWhile are examples. A stateless operation (map, filter) looks at one element at a time. A stateful operation (sorted, distinct) must remember earlier elements, and sorted must read every element before it emits the first one.

Java

import java.util.stream.IntStream;

public class ShortCircuit {
    public static void main(String[] args) {
        int[] seen = {0};
        int first = IntStream.rangeClosed(1, 1_000_000)
                .peek(i -> seen[0]++)
                .filter(i -> i % 7 == 0)
                .findFirst()
                .orElseThrow();
        System.out.println("first = " + first + ", elements read = " + seen[0]);

        seen[0] = 0;
        int firstSorted = IntStream.rangeClosed(1, 1_000)
                .peek(i -> seen[0]++)
                .boxed()
                .sorted((a, b) -> b - a)
                .findFirst()
                .orElseThrow();
        System.out.println("first after sorted = " + firstSorted + ", elements read = " + seen[0]);

        System.out.println(IntStream.of(1, 2, 3, 10, 4).takeWhile(i -> i < 5).boxed().toList());
        System.out.println(IntStream.of(1, 2, 3, 10, 4).dropWhile(i -> i < 5).boxed().toList());
    }
}

Output

first = 7, elements read = 7
first after sorted = 1000, elements read = 1000
[1, 2, 3]
[10, 4]

findFirst stops after 7 elements. With sorted in the pipeline, all 1,000 elements are read, because the largest one could be last. peek runs an action on each element as it passes and exists for debugging. The one-element array seen is a counter the lambda can update: a lambda cannot assign a captured local variable, but it can change an element of a captured array. takeWhile stops at the first element that fails the test; dropWhile skips elements until the first failure and keeps the rest.

Example 3: `map`, `flatMap`, `mapMulti`, and `toList`

map turns each element into one element. flatMap turns each element into a stream and joins those streams. mapMulti (Java 16) does the same job by pushing zero or more elements into a consumer, so no stream is created per element. toList (Java 16) collects into an unmodifiable list.

Java

import java.util.List;
import java.util.stream.Collectors;
import java.util.stream.Stream;

public class MapFlatMapMulti {
    record Order(String id, List<String> symbols) {}

    public static void main(String[] args) {
        List<Order> orders = List.of(
                new Order("o1", List.of("AAPL", "MSFT")),
                new Order("o2", List.of()),
                new Order("o3", List.of("IBM")));

        System.out.println(orders.stream().map(Order::id).toList());

        System.out.println(orders.stream()
                .flatMap(o -> o.symbols().stream().map(s -> o.id() + ":" + s))
                .toList());

        List<String> lines = orders.stream()
                .<String>mapMulti((order, sink) -> {
                    for (String symbol : order.symbols()) {
                        sink.accept(order.id() + ":" + symbol);
                    }
                })
                .toList();
        System.out.println(lines);

        List<Integer> unmodifiable = Stream.of(3, 1, 2).sorted().toList();
        List<Integer> mutable = Stream.of(3, 1, 2).sorted().collect(Collectors.toList());
        mutable.add(4);
        System.out.println(unmodifiable + " " + mutable);
        try {
            unmodifiable.add(4);
        } catch (UnsupportedOperationException e) {
            System.out.println("toList result is unmodifiable");
        }
        System.out.println(Stream.of("a", null).toList());
    }
}

Output

[o1, o2, o3]
[o1:AAPL, o1:MSFT, o3:IBM]
[o1:AAPL, o1:MSFT, o3:IBM]
[1, 2, 3] [1, 2, 3, 4]
toList result is unmodifiable
[a, null]

The type witness <String> on mapMulti is often needed, because the compiler cannot infer the output type from a lambda that only calls sink.accept. Collectors.toList() makes no promise about mutability; it returns an ArrayList in current JDKs. Stream.toList() accepts null elements, unlike List.of and List.copyOf.

Example 4: `reduce` and `collect`

reduce combines all elements into one value with an identity (the start value) and an associative function (one where (a op b) op c equals a op (b op c)). collect performs a mutable reduction: it adds elements into a container such as a list or a StringBuilder.

Java

import java.util.ArrayList;
import java.util.List;
import java.util.Optional;
import java.util.stream.Stream;

public class ReduceAndCollect {
    public static void main(String[] args) {
        List<Integer> qty = List.of(5, 3, 8, 1);

        int total = qty.stream().reduce(0, Integer::sum);
        Optional<Integer> max = qty.stream().reduce(Integer::max);
        Optional<Integer> none = Stream.<Integer>empty().reduce(Integer::max);
        System.out.println("total = " + total + ", max = " + max.orElseThrow()
                + ", empty max present = " + none.isPresent());

        int totalLength = Stream.of("ab", "cde", "f").reduce(0, (sum, s) -> sum + s.length(), Integer::sum);
        System.out.println("total length = " + totalLength);

        ArrayList<Integer> copy = qty.stream().collect(ArrayList::new, ArrayList::add, ArrayList::addAll);
        StringBuilder sb = qty.stream().collect(StringBuilder::new, StringBuilder::append, StringBuilder::append);
        System.out.println(copy + " " + sb);
    }
}

Output

total = 17, max = 8, empty max present = false
total length = 6
[5, 3, 8, 1] 5381

The one-argument reduce returns an Optional because an empty stream has no result. The three-argument form takes an accumulator that changes the type (Integer plus String gives Integer) and a combiner that merges two partial results; the combiner is used when a stream is split for parallel work, covered in Gatherers, Parallel Streams, and Stream Cost. collect takes a supplier for a new container, an accumulator that adds one element, and a combiner that merges two containers.

Example 5: `Collectors` for grouping and maps

Collectors holds ready-made collectors. groupingBy builds a map from a key to the elements with that key, and a second downstream collector decides what each group becomes. partitioningBy splits by a predicate into true and false groups. toMap needs a merge function when keys repeat.

Java

import java.util.List;
import java.util.Map;
import java.util.TreeMap;
import java.util.stream.Collectors;

public class GroupingCollectors {
    record Trade(String symbol, String side, int qty) {}

    public static void main(String[] args) {
        List<Trade> trades = List.of(
                new Trade("AAPL", "BUY", 100), new Trade("MSFT", "SELL", 50),
                new Trade("AAPL", "SELL", 30), new Trade("IBM", "BUY", 70),
                new Trade("MSFT", "BUY", 20));

        Map<String, Long> countBySymbol = trades.stream()
                .collect(Collectors.groupingBy(Trade::symbol, TreeMap::new, Collectors.counting()));
        System.out.println(countBySymbol);

        Map<String, Integer> qtyBySymbol = trades.stream()
                .collect(Collectors.groupingBy(Trade::symbol, TreeMap::new,
                        Collectors.summingInt(Trade::qty)));
        System.out.println(qtyBySymbol);

        Map<Boolean, List<String>> bigTrades = trades.stream()
                .collect(Collectors.partitioningBy(t -> t.qty() >= 50,
                        Collectors.mapping(Trade::symbol, Collectors.toList())));
        System.out.println(bigTrades);

        try {
            trades.stream().collect(Collectors.toMap(Trade::symbol, Trade::qty));
        } catch (IllegalStateException e) {
            System.out.println(e.getMessage());
        }
        Map<String, Integer> merged = trades.stream()
                .collect(Collectors.toMap(Trade::symbol, Trade::qty, Integer::sum, TreeMap::new));
        System.out.println(merged);
    }
}

Output

{AAPL=2, IBM=1, MSFT=2}
{AAPL=130, IBM=70, MSFT=70}
{false=[AAPL, MSFT], true=[AAPL, MSFT, IBM]}
Duplicate key AAPL (attempted merging values 100 and 30)
{AAPL=130, IBM=70, MSFT=70}

The TreeMap::new argument picks the map type, so the keys print sorted; the default groupingBy map is a HashMap, whose order is not specified. toMap without a merge function throws IllegalStateException on the first duplicate key. partitioningBy always returns both keys, even when a group is empty.

Example 6: `joining`, `teeing`, and primitive streams

joining concatenates strings with a delimiter, prefix, and suffix. teeing (Java 12) sends every element to two collectors and merges their results. Primitive streams (IntStream, LongStream, DoubleStream) hold int, long, or double values without boxing and add sum, average, and summaryStatistics.

Java

import java.util.IntSummaryStatistics;
import java.util.List;
import java.util.stream.Collectors;
import java.util.stream.IntStream;

public class JoiningTeeingPrimitive {
    record Range(int min, int max) {}

    public static void main(String[] args) {
        List<String> symbols = List.of("AAPL", "MSFT", "IBM");
        System.out.println(symbols.stream().collect(Collectors.joining(", ", "[", "]")));

        List<Integer> prices = List.of(190, 410, 90, 300);
        Range range = prices.stream().collect(Collectors.teeing(
                Collectors.minBy(Integer::compare),
                Collectors.maxBy(Integer::compare),
                (min, max) -> new Range(min.orElseThrow(), max.orElseThrow())));
        System.out.println(range);

        int sum = prices.stream().mapToInt(Integer::intValue).sum();
        IntSummaryStatistics stats = prices.stream().mapToInt(Integer::intValue).summaryStatistics();
        System.out.println("sum = " + sum + ", " + stats);
        System.out.println(IntStream.range(0, 5).average().orElseThrow());
        System.out.println(IntStream.rangeClosed(1, 3).mapToObj(i -> "row" + i).toList());
    }
}

Output

[AAPL, MSFT, IBM]
Range[min=90, max=410]
sum = 990, IntSummaryStatistics{count=4, sum=990, min=90, average=247.500000, max=410}
2.0
[row1, row2, row3]

mapToInt converts a Stream<Integer> into an IntStream; mapToObj and boxed go back. range excludes the end value and rangeClosed includes it. average returns an OptionalDouble because an empty stream has no average.

Example 7: infinite streams, ordering, and side effects

Stream.iterate and Stream.generate create infinite streams, which need a short-circuiting operation such as limit to end. A stream from a List is ordered (it has an encounter order); a stream from a HashSet is not. Side effects in lambdas, such as adding to an outside list, are allowed by forEach but break when the stream runs in parallel.

Java

import java.util.ArrayList;
import java.util.List;
import java.util.function.Supplier;
import java.util.stream.Stream;

public class InfiniteAndOrdering {
    public static void main(String[] args) {
        System.out.println(Stream.iterate(1, x -> x * 2).limit(6).toList());
        System.out.println(Stream.iterate(1, x -> x < 100, x -> x * 3).toList());

        int[] next = {0};
        Supplier<Integer> ids = () -> ++next[0];
        System.out.println(Stream.generate(ids).limit(3).toList());

        List<Integer> source = List.of(5, 1, 4, 2, 3);
        System.out.println(source.stream().filter(i -> i > 1).toList());

        List<Integer> sideEffect = new ArrayList<>();
        source.stream().filter(i -> i % 2 == 1).forEach(sideEffect::add);
        List<Integer> noSideEffect = source.stream().filter(i -> i % 2 == 1).toList();
        System.out.println(sideEffect + " " + noSideEffect);
    }
}

Output

[1, 2, 4, 8, 16, 32]
[1, 3, 9, 27, 81]
[1, 2, 3]
[5, 4, 2, 3]
[5, 1, 3] [5, 1, 3]

The three-argument iterate (Java 9) works like a for loop: seed, condition, next value. filter keeps the encounter order of the list. Both odd-number lists are equal here, but the toList version has no shared mutable state, so it stays correct when the stream is made parallel; the forEach(sideEffect::add) version does not, because ArrayList is not thread-safe.

Example 8: a stream is one-shot, and a realistic report

A stream can be consumed only once. A second terminal operation throws IllegalStateException. A Supplier<Stream<T>> creates a fresh stream each time it is needed. The final program combines the earlier operations into a position report.

Java

import java.util.List;
import java.util.Map;
import java.util.TreeMap;
import java.util.function.Supplier;
import java.util.stream.Collectors;
import java.util.stream.Stream;

public class PositionReport {
    record Fill(String account, String symbol, String side, int qty, long priceCents) {
        long signedQty() { return side.equals("BUY") ? qty : -qty; }
        long notional() { return qty * priceCents; }
    }

    public static void main(String[] args) {
        List<Fill> fills = List.of(
                new Fill("A1", "AAPL", "BUY", 100, 19_000),
                new Fill("A1", "AAPL", "SELL", 40, 19_100),
                new Fill("A2", "MSFT", "BUY", 10, 41_000),
                new Fill("A1", "IBM", "BUY", 5, 9_000),
                new Fill("A2", "MSFT", "SELL", 10, 41_200),
                new Fill("A2", "AAPL", "BUY", 20, 19_050));

        Stream<Fill> once = fills.stream();
        once.count();
        try {
            once.count();
        } catch (IllegalStateException e) {
            System.out.println(e.getMessage());
        }

        Supplier<Stream<Fill>> fresh = fills::stream;
        Map<String, Map<String, Long>> positions = fresh.get().collect(Collectors.groupingBy(
                Fill::account, TreeMap::new,
                Collectors.groupingBy(Fill::symbol, TreeMap::new, Collectors.summingLong(Fill::signedQty))));
        System.out.println("positions " + positions);

        Map<String, Long> notionalByAccount = fresh.get().collect(Collectors.groupingBy(
                Fill::account, TreeMap::new, Collectors.summingLong(Fill::notional)));
        System.out.println("notional cents " + notionalByAccount);

        String flat = positions.entrySet().stream()
                .flatMap(a -> a.getValue().entrySet().stream()
                        .filter(p -> p.getValue() == 0)
                        .map(p -> a.getKey() + "/" + p.getKey()))
                .collect(Collectors.joining(", "));
        System.out.println("flat positions: " + flat);
    }
}

Output

stream has already been operated upon or closed
positions {A1={AAPL=60, IBM=5}, A2={AAPL=20, MSFT=0}}
notional cents {A1=2709000, A2=1203000}
flat positions: A2/MSFT

The nested groupingBy builds a map of maps: account, then symbol, then net quantity. Prices are in cents as long, so sums are exact. The flat-position search streams over the entries of the result map; a Map is not a stream source itself, but its entrySet() is a collection.

Interview points

A pipeline is a source, intermediate operations, and one terminal operation; nothing runs until the terminal operation starts.
Elements flow one at a time through all stages; sorted and other stateful operations break this by buffering.
Short-circuiting operations (findFirst, anyMatch, limit, takeWhile) make infinite streams usable.
mapMulti avoids a stream per element compared with flatMap; it often needs a type witness.
Stream.toList() is unmodifiable and allows null; Collectors.toList() makes no mutability promise.
reduce needs an associative function and a true identity; collect is the mutable reduction.
toMap throws IllegalStateException on duplicate keys unless given a merge function.
groupingBy defaults to a HashMap; pass a map factory for a sorted result.
Side effects in stream lambdas break under parallel execution; collect results instead.
A stream is one-shot; use a Supplier<Stream<T>> to get a new one.

Summary

Intermediate operations are lazy; terminal operations start the work.
Stateless operations look at one element; stateful ones (sorted, distinct) keep state.
map is one to one, flatMap and mapMulti are one to many.
Collectors.groupingBy, partitioningBy, toMap, joining, and teeing cover most aggregation.
IntStream, LongStream, and DoubleStream avoid boxing and add sum, average, and statistics.
iterate and generate produce infinite streams that need limit or takeWhile.
List sources are ordered; HashSet sources are not.
A consumed stream throws IllegalStateException on reuse.

Example 1: a pipeline is lazy

Example 2: short-circuiting and stateful operations

Example 3: map, flatMap, mapMulti, and toList

Example 4: reduce and collect

Example 5: Collectors for grouping and maps

Example 6: joining, teeing, and primitive streams

Example 7: infinite streams, ordering, and side effects

Example 8: a stream is one-shot, and a realistic report

Interview points

Summary

Example 3: `map`, `flatMap`, `mapMulti`, and `toList`

Example 4: `reduce` and `collect`

Example 5: `Collectors` for grouping and maps

Example 6: `joining`, `teeing`, and primitive streams