When to Stream, Collect, and Gather

Slide navigation: Forward with space bar, →, or swipe left. Backwards with ← or swipe right. Switch slides with PgDn and PgUp.

Slides at https://horstmann.com/presentations/2024/baselone

About the Speaker

• Cay Horstmann
• Author of Core Java (13 editions since 1996), Java/JavaScript/Scala for the Impatient, etc.
• https://horstmann.com

Why Use Streams?

• Can be parallelized
• Can be more efficient (no intermediate collections)
• Declarative (“what, not how”)

  txns.stream()
      .filter(t -> t.getBuyer().getAge() >= 65)
      .map(Txn::getSeller)
      .distinct()
      .sorted(comparing(Seller::getName))
      .map(Seller::getName)
      .forEach(System.out::println);
  

• instead of

  Set<Seller> sellers = new HashSet<>();
  for (Txn t : txns) {
      if (t.getBuyer().getAge() >= 65)
          sellers.add(t.getSeller());
  }
  List<Seller> sorted = new ArrayList<>(sellers);
  Collections.sort(sorted, new Comparator<Seller>() {
      public int compare(Seller a, Seller b) {
          return a.getName().compareTo(b.getName());
      }
  });
  for (Seller s : sorted)
      System.out.println(s.getName());
  

  (Source)

Why Not Always Streams?

  static int numberOfBrickWithChildrenWithAtLeastTwoParents(String input) {
    var grid = new HashMap<Pos, Level>();
    return input.lines()
        .map(l -> {
          var array = Arrays.stream(l.split(",|~")).mapToInt(Integer::parseInt).toArray();
          return new Brick(array[0], array[3], array[1], array[4], array[2], array[5]);
        })
        .collect(groupingBy(Brick::z1, TreeMap::new, toList()))
        .values().stream()
        .flatMap(List::stream)
        .map(brick -> {
          var onTop = brick.positions()
              .map(pos -> grid.getOrDefault(pos, new Level(null, 0)))
              .collect(groupingBy(Level::z, TreeMap::new, flatMapping(l -> Stream.ofNullable(l.brick), toSet())))
              .lastEntry();
          var level =  new Level(brick, 1 + brick.z2 - brick.z1 + onTop.getKey());
          brick.positions().forEach(pos -> grid.put(pos, level));
          return new Pair<>(brick, onTop.getValue());
        })
        .collect(teeing(
            toMap(Pair::u, pair -> pair.v().size()),
            flatMapping(pair -> pair.v().stream().map(b -> new Pair<>(b, pair.u())),
                groupingBy(Pair::u, mapping(Pair::v, toList()))),
            (parentMap, childrenMap) ->
                (int) parentMap.keySet().stream()
                    .filter(parent -> childrenMap.getOrDefault(parent, List.of()).stream()
                        .allMatch(child -> parentMap.get(child) > 1))
                    .count()
        ));
  }

Wer misst, misst Mist

• JMH benchmark with pre-initialized Point[] state.randoms, comparing

  ArrayList<Point> points = new ArrayList<>(List.of(state.randoms));
  Point largest = points.get(0);
  for (int i = 1; i < points.size(); i++) {
      Point p = points.get(i);
      if (p.x % 2 == 0 && p.y % 2 == 0) {
          if (p.x * p.x + p.y * p.y > largest.x * largest.x + largest.y * largest.y)
              largest = p;
      }
  }
  

  with

  Point largest = Stream.of(state.randoms)
      .filter(p -> p.x % 2 == 0 && p.y % 2 == 0)
      .max(Comparator.comparingInt(p -> p.x * p.x + p.y * p.y))
      .orElseThrow();
  

• Streams are twice as fast!

  MyBenchmark.withCollection  avgt  100  206.1 ± 8.9 ms/op
  MyBenchmark.withStream      avgt  100   89.3 ± 0.6 ms/op
  

• Ok, that was unfair. Using

  ArrayList<Point> points = new ArrayList<>(List.of(state.randoms));
  ...points.stream()...
  

  yields

  MyBenchmark.withCollection  avgt  100  199.2 ± 8.9 ms/op
  MyBenchmark.withStream      avgt  100  212.6 ± 8.7 ms/op
  

• Benchmarks such as this one by Nikolai Parlog indicate that the stream overhead is minimal
• For large data, also consider memory consumption

Max and Min

• Classic max/min:

  T largest = Collections.max(collection, comparator);
  

• Stream version:

  T largest = collection.stream().max(comparator).orElse(default);
  

• Stream version behaves better with empty collection
• Stream version works with array
• Verdict: In 2024, use streams for max/min

Finding, Counting, Collecting Matches

• Can find element in a list with indexOf
• But only if match means equals
• General loop is messy:

  boolean found = false;
  int index = 0; 
  while (index < list.length() && !found) {
     if (list.get(index) is a match) found = true;
     else index++;
  }
  if (found) ... 
  

• Streams are a big win:

  Optional<T> result = list.stream().filter(e -> e is a match).findFirst();
  

• Counting and collecting all matches are also simple:

  long count = list.stream().filter(e -> e is a match).count();
  List<T> matches = list.stream().filter(e -> e is a match).toList();
  

Nothing to See Here...

• Advent of Code 2023 Problem 1: Add the numbers made from the first and last digit in each input line:

  1abc2
  pqr3stu8vwx
  a1b2c3d4e5f
  

• Classic loop solution:

  Pattern p = Pattern.compile("^[^\\d]*(\\d).*(\\d)[^\\d]*$");
  List<String> lines = Files.readAllLines(path);
  int sum = 0;
  for (String line : lines) {
      Matcher m = p.matcher(line);
      if (m.matches()) {
          int firstLast = Integer.parseInt(m.group(1) + m.group(2));
          sum += firstLast;
      }
  }
  

• Stream solution:

  int sum = Files.lines(path)
      .map(line -> p.matcher(line))
      .filter(m -> m.matches())
      .mapToInt(m -> Integer.parseInt(m.group(1) + m.group(2)))
      .sum();
  

• Stream solution sends crucial message—move along, nothing to see here...

Method Expressions

• Previous solution with method expressions:

  int sum = lines.stream()
      .map(p::matcher) // Instead of line -> p.matcher(line)
      .filter(Matcher::matches) // Instead of m -> m.matches()
      .mapToInt(m -> Integer.parseInt(m.group(1) + m.group(2)))
      .sum();
  

• Helpful table

  Variant	Lambda expression	Method expression
  Class::staticMethod	x -> Pattern.compile(x)	Pattern::compile
  Class::instanceMethod	x -> x.matches()	Matcher::matches
  object::instanceMethod	x -> p.matcher(x)	p::matcher
  Out of luck	x -> x.substring(k)

• Verdict: Use them—the method name is clearer than the method body.

For Each

• Oracle Tutorial by Venkat Subramanian:

  for (int i = 0; i < 5; i++) {
    System.out.println(i);
  }
  

  ⇒

  IntStream.range(0, 5)
    .forEach(i -> System.out.println(i));
  

• Steps:

  for (int i = 0; i < 15; i += 3) ...
  
  

  ⇒

  IntStream.iterate(0, i -> i < 15, i -> i + 3).forEach(...)
  

• Nested loop:

  for (int i = 0; i < 5; i++)
     for (int j = 0; j < 10; j++)
        System.out.println(i + " " + j);
  

  ⇒

  IntStream.range(0, 5).forEach(i ->
      IntStream.range(0, 10).forEach(j ->
          System.out.println(i + " " + j)))
  

• Not easier to read
• Can't mutate variables in lambda
• Not really “functional”
• Verdict: Avoid forEach

Streams are Everywhere in the Java API

• Lines: String.lines(), Files.lines(path), HttpResponse.BodyHandlers.ofLines()
• Tokens: Scanner.tokens(), Pattern.splitAsStream(input)
• Matches: Matcher.results(), Scanner.findAll(pattern)
• Files: Files.list(path), Files.walk(path), Files.find(...)
• Process handles: Process.children(), ProcessHandle.allProcesses(), ...
• Dates: LocalDate.datesUntil(endExclusive)
• Random numbers: RandomGenerator.ints(), RandomGenerator.ints(from, to)
• Code points and code units: String.codePoints(), String.chars()
• Miscellaneous: BitSet.stream(), Locale.availableLocales(), NetworkInterface.inetAddresses(), DriverManager.drivers()
• More attractive than returning a collection or iterator
• Particularly for lazy evaluation
• Nudges you into the stream API
• You always have the option to call toList() or limit(n).toList()
• Or, ugh, forEach

Complex Lambdas

• Advent of Code 2023 Day 2: Given upper limit (only 12 red, 13 green, and 14 blue) and

  Game 1: 3 blue, 4 red; 1 red, 2 green, 6 blue; 2 green
  Game 2: 1 blue, 2 green; 3 green, 4 blue, 1 red; 1 green, 1 blue
  Game 3: 8 green, 6 blue, 20 red; 5 blue, 4 red, 13 green; 5 green, 1 red
  ...
  

  add all “possible” game IDs
• Rémi's ingenious solution:

  var expected = Map.of("red", 12, "green", 13, "blue", 14);
  var PATTERN = Pattern.compile("Game ([0-9]+)|([0-9]+) (red|green|blue)");
  var result = input.lines()
      .mapToInt(line -> {
          var matcher = PATTERN.matcher(line);
          matcher.find();
          var gameId = parseInt(matcher.group(1));
          return matcher.results().allMatch(result -> parseInt(result.group(2)) <= expected.get(result.group(3))) ? gameId : 0;
      })
      .sum();
  

• Avoid lambdas with complex bodies
• Hard to read, hard to debug
• Use named method:

  .mapToInt(line -> gameIdIfPossible(line))
  .sum()
  

• Break into smaller steps:

  .map(Game::parse)
  .filter(Game::isPossible)
  .mapToInt(Game::id)
  .sum()
  

Nested Streams

• Skipping empty optionals with flatMap and stream:

  var commandLines = ProcessHandle.allProcesses()
      .flatMap(p -> p.info().commandLine().stream())
      .toList()
  

• Rows and columns with flatMap and map:

  var result = IntStream.range(0, 5).boxed().flatMap(i ->
          IntStream.range(0, 10).mapToObj(j ->
              "" + i + " " + j))
      .toList(); // ["0 0", "0 1", ..., "4 9"]
  

• Same tips from previous slide apply:

  // Advent of Code 2023 Day 9
  return input.lines()
      .map(line -> Arrays.stream(line.split(" ")).map(Integer::parseInt).toList())
      .mapToInt(history -> Stream.iterate(history,
              h -> !h.stream().allMatch(v -> v == 0),
              h -> range(1, h.size()).mapToObj(i -> h.get(i) - h.get(i - 1)).toList())
          .mapToInt(List::getLast)
          .sum())
      .sum();
  

• Break into named tasks instead

Ad-Hoc Records

• This just in from Devoxx Belgium 2024

  record IndexWord(int index, String value) {}
  
  IndexWord splitStream(String line) {
      var pattern = Pattern.compile("...");
      var words = Pattern.split(line);
      return IntStream.range(0, words.length)
          .filter(i -> words[i].length() == 3)
          .mapToObj(i -> new IndexWord(i, words[i]))
          .findFirst().orElseThrow();
  }
  

• Ugh—a whole new type for this use case
• Consider

  record<T> IndexedValue(int index, T value) {}
  

• Or use Map.Entry<K, V> as poor man's pair
• Ad-hoc records are fine when they are local
• Why not use

  new Scanner(line).useDelimiter("...").tokens()
  

• Hold that thought—need zipWithIndex

Collectors

• Terminal stream operation collect(Collector)
• Collectors provides 40 methods that yield implementations
• Can also roll your own (uncommon)
• Can be easy:

  Map<Integer, String> idToName = people.collect(
      Collectors.toMap(Person::id, Person::name));
  

• Or baffling:

  Map.Entry<List<String>, Double> result = // Map.Entry is poor man's Pair
      Stream.of(cities)
      .filter(c -> c.state().equals("NV"))
      .collect(teeing(
          mapping(City::name, toList()), // First downstream collector
          averagingDouble(City::population), // Second downstream collector
          Map::entry)); // Combining function
  

When to Collect

• By far the most common case: Collecting to a map

  Map<Integer, String> idToName = people.collect(toMap(Person::id, Person::name));
      // Only use when the keys are distinct
  Map<String, List<Person>> nameToPerson = people.collect(groupingBy(Person::name));
  

• Tip: import static java.util.stream.Collectors.*;
• Map of counts:

  Map<String, Long> counts = Stream.of(words)
      .collect(groupingBy(s -> s, counting()));
  

• Why not Function::identity instead of s -> s?
  • Are you kidding? That's fourteen more characters!
  • Error:

    |  no suitable method found for groupingBy(Function::identity,java.util.stream.Collector<java.lang.Object,capture#9 of ?,java.lang.Long>)
    |      method java.util.stream.Collectors.<T,K,D,A,M>groupingBy(java.util.function.Function<? super T,? extends K>,java.util.function.Supplier<M>,java.util.stream.Collector<? super T,A,D>) is not applicable
    |        (cannot infer type-variable(s) T,K,D,A,M
    |          (actual and formal argument lists differ in length))
    |      method java.util.stream.Collectors.<T,K,A,D>groupingBy(java.util.function.Function<? super T,? extends K>,java.util.stream.Collector<? super T,A,D>) is not applicable
    |        (cannot infer type-variable(s) T,K,A,D,capture#10 of ?,T
    |          (argument mismatch; unexpected static method <T>identity() found in unbound lookup))
    |      method java.util.stream.Collectors.<T,K>groupingBy(java.util.function.Function<? super T,? extends K>) is not applicable
    |        (cannot infer type-variable(s) T,K
    |          (actual and formal argument lists differ in length))
    |      .collect(groupingBy(Function::identity, counting()));
    |               ^--------^
    
    

  • Oops—that should be Function.identity()
• That's probably clearer than the classic loop:

  for (String word : words) counts.merge(word, 1L, Long::sum);
  

• Set-valued map:

  var index = Stream.of(indexEntries).collect(
      groupingBy(Entry::term,
          mapping(Entry::page, 
              toCollection(TreeSet::new)))); // toSet() yields a hash set
  

• Classic solution:

  var index = new TreeMap<String, TreeSet<Integer>>();
  for (var e : indexEntries)
     index.computeIfAbsent(e.term(), _ -> new TreeSet<>()).add(e.page());
  

• What do you find clearer?

Gatherers

• JEP 485 introduces gather for intermediate operations
• Five built-in gatherers
• Fixed and sliding window:

  IntStream.range(0, 10).boxed().gather(Gatherers.windowFixed(4)).toList()
      // [[0, 1, 2, 3], [4, 5, 6, 7], [8, 9]]
  IntStream.range(0, 10).boxed().gather(Gatherers.windowSliding(4)).toList()
      // [[0, 1, 2, 3], [1, 2, 3, 4], [2, 3, 4, 5], [3, 4, 5, 6], [4, 5, 6, 7], [5, 6, 7, 8], [6, 7, 8, 9]]
  

• Useful to compare/combine/group adjacent elements
• Scan and fold:

  Integer[] digits = { 1, 7, 2, 9 };
  Stream.of(digits).gather(Gatherers.scan(() -> 0, (x, y) -> x * 10 + y)) // [1, 17, 172, 1729]
  Stream.of(digits).gather(Gatherers.fold(() -> 0, (x, y) -> x * 10 + y)) // [1729]
  

• mapConcurrent—see below
• Rolling your own is cumbersome:

  static <T> Gatherer<T, ?, Map.Entry<Long, T> zipWithIndex() {
      class Index { long index = 0; }
      return Gatherer.ofSequential(
              Index::new,
              Gatherer.Integrator.ofGreedy((state, element, downstream) ->
                      downstream.push(Map.Entry(state.index++, element))));
  }
  

• Verdict: Nice, but missing a few gatherers (zip, zipWithIndex)

Checked Exceptions

• JavaSpecialists newsletter #314:

  Files.walkFileTree(source, new SimpleFileVisitor<>() {
      public FileVisitResult visitFile(Path file, BasicFileAttributes attrs) throws IOException {
          if (Files.isSymbolicLink(file)) {
              Path targetLink = target.resolve(source.relativize(file));
              Files.createDirectories(targetLink.getParent());
              Files.copy(file, targetLink, StandardCopyOption.REPLACE_EXISTING, LinkOption.NOFOLLOW_LINKS);
          }
          return FileVisitResult.CONTINUE;
  }});

• I chide Heinz for not using modern I/O. So much simpler. Except...exceptions...

      try (Stream<Path> walk = Files.walk(source)) {
          walk.forEach(file -> {
              if (Files.isSymbolicLink(file)) {
                  Path targetLink =target.resolve(source.relativize(file));
                  try {
                      Files.createDirectories(targetLink.getParent());
                      Files.copy(file, targetLink, StandardCopyOption.REPLACE_EXISTING, LinkOption.NOFOLLOW_LINKS);
                  } catch (IOException e) {
                      throw new UncheckedIOException(e);
                  }
              }
          });
      }
      catch (UncheckedIOException e) {
           throw e.getCause();
      }
  

Checked Exceptions

• Standard advice: Use helper method
• Apache commons.lang FailableStream
• sneakyfun, Lombok @SneakyThrow

  sneaky(targetLink -> {
      Files.createDirectories(targetLink.getParent());
      Files.copy(file, targetLink, StandardCopyOption.REPLACE_EXISTING, LinkOption.NOFOLLOW_LINKS);
  });
  

• Custom gatherer (for map/filter)

  interface CheckedFunction<T, R> { R apply(T arg) throws Exception; }
  
  <T, R> Gatherer<T, ?, R> checkedMap(CheckedFunction<? super T,? extends R> mapper) {
      return Gatherer.<T, R>of(
          Gatherer.Integrator.ofGreedy((_, element, downstream) -> {
              try {
                  return downstream.push(mapper.apply(element));
              } catch (Exception ex) {
                  throw new RuntimeException("mapChecked", ex);
              }
          }));
  }
  

• or collector (for forEach):

  walk.collect(checkedForEach(file -> ...))
  

Concurrency

• Parallel streams are great for processor-intensive tasks:

  var results = collection
      .parallelStream()
      .map(e -> hardWork(e))
      .toList();
  

• Stream splits into n subranges (n = number of cores)
• Subrange tasks execute on global fork-join pool
• Results are assembled in order
• Expect n-fold speedup
• Does not work well for tasks that mostly block
• Could have much higher throughput with another executor (thread pool, virtual threads)
• Well-known workaround does not work with those executors
• Gatherers to the rescue:

  int MAX_CONCURRENT = 1000;
  var results = collection
      .stream() // Not parallelStream!
      .gather(Gatherers.mapConcurrent(MAX_CONCURRENT, e -> blockingWork(e)))
      .toList();
  

Demo: Phone Mnemonics

• Given a phone number, what word or word sequence spells it on a touch-tone pad?
• For example, 435 569 6753 → "HELLO WORLD" (and maybe others)
• From Martin Odersky's 2011 State of Scala, adapted from this paper by Lutz Prechelt
• Solution follows my 2015 Scala course
• 68 lines of code in JTaccuino notebook
• That's 75% less than the median for Java in Prechelt's paper
• For better or worse, no loops, no recurson. Only streams (and a custom gatherer)
• The tempation to nest streams is irresistible
• No fun: Fixing compiler errors, debugging nested streams
• Pretty opaque even after cleanup
• Would need another round of refactoring
• Verdict: Until we all get more fluent with flatMap and fold, budget time for refactoring and documentation

The End

• Streams and loop performance are on par (unless you do something wasteful)
• Favor streams when they are easier to understand than equivalent loops
• Good use cases:
  • Max, min, sorting
  • sums and averages
  • finding, counting, and collecting matches
• Favor method expressions
• Complex lambdas, nested streams, are not nice. Refactor into helper methods.
• “Functional” is not pixie dust. Don't turn every loop into a stream.
• Checked exceptions in stream lambdas are a pain. Know your remedies.
• Embrace simple idioms for collecting into maps
• Don't go wild with downstream collectors
• There are a handful of useful gatherers
• Budget time for refactoring

Questions?