Lab 13: Streams & Lambdas

Objective

Use Java Streams API for data processing — filter, map, reduce, collect, flatMap, and parallel streams — and write concise lambda expressions and method references.

Background

The Streams API (Java 8+) brings functional-style data processing to Java. A Stream is a lazy sequence of elements supporting aggregate operations. Streams enable declarative, pipeline-based data transformation that replaces verbose for-loop boilerplate. Combined with lambdas and method references, they are the most impactful Java 8 feature for day-to-day code.

Time

45 minutes

Prerequisites

• Lab 08 (Interfaces — Functional Interfaces)

• Lab 09 (Collections)

• Lab 12 (Generics)

Tools

• Java 21 (Eclipse Temurin)

• Docker image: innozverse-java:latest

---

Lab Instructions

Step 1: Stream Pipeline Basics

// StreamBasics.java
import java.util.*;
import java.util.stream.*;

public class StreamBasics {
    public static void main(String[] args) {
        List<Integer> numbers = List.of(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);

        // filter → map → collect
        List<Integer> result = numbers.stream()
            .filter(n -> n % 2 == 0)      // [2,4,6,8,10]
            .map(n -> n * n)               // [4,16,36,64,100]
            .collect(Collectors.toList()); // terminal
        System.out.println("Even squares: " + result);

        // toList() — unmodifiable (Java 16+)
        var squares = numbers.stream().map(n -> n * n).toList();
        System.out.println("All squares: " + squares);

        // forEach — terminal, void
        System.out.print("Odds: ");
        numbers.stream().filter(n -> n % 2 != 0).forEach(n -> System.out.print(n + " "));
        System.out.println();

        // count, min, max, sum, average
        long count = numbers.stream().filter(n -> n > 5).count();
        Optional<Integer> max = numbers.stream().max(Integer::compareTo);
        int sum = numbers.stream().mapToInt(Integer::intValue).sum();
        OptionalDouble avg = numbers.stream().mapToInt(Integer::intValue).average();

        System.out.printf("%ncounting >5: %d, max: %d, sum: %d, avg: %.1f%n",
            count, max.get(), sum, avg.getAsDouble());

        // Stream.of, Stream.iterate, Stream.generate
        Stream.of("a", "b", "c").forEach(System.out::print);
        System.out.println();

        Stream.iterate(0, n -> n + 3).limit(5).forEach(n -> System.out.print(n + " "));
        System.out.println();

        Stream.generate(Math::random).limit(3)
            .map(d -> String.format("%.3f", d)).forEach(s -> System.out.print(s + " "));
        System.out.println();
    }
}

💡 Streams are lazy — intermediate operations (filter, map) don't execute until a terminal operation (collect, forEach, count) is called. This enables short-circuit optimization: stream().filter(...).findFirst() stops at the first match without processing the rest.

📸 Verified Output:

Even squares: [4, 16, 36, 64, 100]
All squares: [1, 4, 9, 16, 25, 36, 49, 64, 81, 100]
Odds: 1 3 5 7 9

counting >5: 5, max: 10, sum: 55, avg: 5.5
abc
0 3 6 9 12
0.472 0.651 0.203

---

Step 2: Lambdas & Method References

// LambdasMethodRefs.java
import java.util.*;
import java.util.function.*;
import java.util.stream.*;

public class LambdasMethodRefs {
    static int triple(int n) { return n * 3; }
    static boolean isEven(int n) { return n % 2 == 0; }

    record Person(String name, int age) {
        String greeting() { return "Hello, I'm " + name; }
    }

    public static void main(String[] args) {
        List<Integer> nums = List.of(1, 2, 3, 4, 5, 6);

        // Lambda forms
        nums.stream().map(n -> n * 2).forEach(n -> System.out.print(n + " "));        // full lambda
        System.out.println();
        nums.stream().map(n -> { int r = n * 2; return r; }).forEach(n -> System.out.print(n + " ")); // block lambda
        System.out.println();

        // Method references — 4 kinds
        // 1. Static method ref
        nums.stream().map(LambdasMethodRefs::triple).forEach(n -> System.out.print(n + " "));
        System.out.println();

        // 2. Instance method ref (of arbitrary object of type)
        List<String> words = List.of("hello", "WORLD", "Java");
        words.stream().map(String::toUpperCase).forEach(s -> System.out.print(s + " "));
        System.out.println();

        // 3. Instance method ref (of particular instance)
        String prefix = "Mr. ";
        words.stream().map(prefix::concat).forEach(s -> System.out.print(s + " "));
        System.out.println();

        // 4. Constructor ref
        List<Person> people = List.of("Alice", "Bob", "Carol").stream()
            .map(name -> new Person(name, 30))
            .toList();
        people.stream().map(Person::greeting).forEach(System.out::println);

        // Filter with method ref
        System.out.println("\nEven (method ref): " + nums.stream().filter(LambdasMethodRefs::isEven).toList());
    }
}

💡 Method references are shorthand for lambdas that just call a method. String::toUpperCase = s -> s.toUpperCase(). System.out::println = x -> System.out.println(x). They're more readable for simple cases and automatically adapt to any functional interface with the right signature.

📸 Verified Output:

2 4 6 8 10 12
2 4 6 8 10 12
3 6 9 12 15 18
HELLO WORLD JAVA
Mr. hello Mr. WORLD Mr. Java
Hello, I'm Alice
Hello, I'm Bob
Hello, I'm Carol

Even (method ref): [2, 4, 6]

---

Step 3: Collectors — Grouping & Partitioning

// Collectors.java
import java.util.*;
import java.util.stream.*;

public class CollectorsDemo {

    record Transaction(String id, String category, double amount, String month) {}

    public static void main(String[] args) {
        List<Transaction> txns = List.of(
            new Transaction("T1", "food",   50.00, "Jan"),
            new Transaction("T2", "tech",  450.00, "Jan"),
            new Transaction("T3", "food",   35.00, "Feb"),
            new Transaction("T4", "tech",  299.00, "Feb"),
            new Transaction("T5", "food",   60.00, "Feb"),
            new Transaction("T6", "travel", 800.00, "Jan"),
            new Transaction("T7", "tech",  120.00, "Mar")
        );

        // groupingBy
        Map<String, List<Transaction>> byCategory =
            txns.stream().collect(Collectors.groupingBy(Transaction::category));
        byCategory.forEach((cat, list) ->
            System.out.printf("%-8s %d transactions%n", cat, list.size()));

        // groupingBy + downstream collector
        System.out.println("\nTotal by category:");
        Map<String, Double> totalByCategory = txns.stream()
            .collect(Collectors.groupingBy(
                Transaction::category,
                Collectors.summingDouble(Transaction::amount)));
        totalByCategory.entrySet().stream()
            .sorted(Map.Entry.<String, Double>comparingByValue().reversed())
            .forEach(e -> System.out.printf("  %-8s $%.2f%n", e.getKey(), e.getValue()));

        // partitioningBy — splits into true/false
        Map<Boolean, List<Transaction>> partition = txns.stream()
            .collect(Collectors.partitioningBy(t -> t.amount() > 100));
        System.out.println("\nHigh value (>$100): " + partition.get(true).size());
        System.out.println("Low value (<=$100): " + partition.get(false).size());

        // joining
        String ids = txns.stream().map(Transaction::id)
            .collect(Collectors.joining(", ", "[", "]"));
        System.out.println("\nIDs: " + ids);

        // counting
        Map<String, Long> countByMonth = txns.stream()
            .collect(Collectors.groupingBy(Transaction::month, Collectors.counting()));
        System.out.println("Per month: " + new TreeMap<>(countByMonth));

        // statistics
        DoubleSummaryStatistics stats = txns.stream()
            .collect(Collectors.summarizingDouble(Transaction::amount));
        System.out.printf("%nStats: count=%d, sum=%.0f, avg=%.0f, min=%.0f, max=%.0f%n",
            stats.getCount(), stats.getSum(), stats.getAverage(), stats.getMin(), stats.getMax());
    }
}

💡 Collectors.groupingBy(classifier, downstream) is incredibly powerful. The downstream collector can be counting(), summingDouble(), joining(), another groupingBy() (nested grouping), or mapping() + toList(). This replaces hundreds of lines of imperative grouping code with a single expression.

📸 Verified Output:

food     3 transactions
tech     3 transactions
travel   1 transactions

Total by category:
  travel   $800.00
  tech     $869.00
  food     $145.00

High value (>$100): 4
Low value (<=$100): 3

IDs: [T1, T2, T3, T4, T5, T6, T7]
Per month: {Feb=3, Jan=3, Mar=1}

Stats: count=7, sum=1814, avg=259, min=35, max=800

---

Step 4: flatMap & Optional

// FlatMapOptional.java
import java.util.*;
import java.util.stream.*;

public class FlatMapOptional {

    record Order(String id, List<String> items) {}

    public static void main(String[] args) {
        // flatMap — flatten nested streams
        List<Order> orders = List.of(
            new Order("O1", List.of("apple", "banana", "cherry")),
            new Order("O2", List.of("apple", "date")),
            new Order("O3", List.of("elderberry", "fig", "apple"))
        );

        // All items across all orders
        List<String> allItems = orders.stream()
            .flatMap(o -> o.items().stream())
            .sorted()
            .toList();
        System.out.println("All items: " + allItems);

        // Unique items
        Set<String> unique = orders.stream()
            .flatMap(o -> o.items().stream())
            .collect(Collectors.toSet());
        System.out.println("Unique: " + new TreeSet<>(unique));

        // Item frequency
        Map<String, Long> freq = orders.stream()
            .flatMap(o -> o.items().stream())
            .collect(Collectors.groupingBy(s -> s, Collectors.counting()));
        System.out.println("Frequency: " + new TreeMap<>(freq));

        // Optional — absence without null
        Optional<String> present = Optional.of("hello");
        Optional<String> empty = Optional.empty();

        System.out.println("\nOptional:");
        System.out.println(present.isPresent());
        System.out.println(present.get().toUpperCase());
        System.out.println(empty.orElse("default"));
        System.out.println(empty.orElseGet(() -> "computed-" + System.currentTimeMillis() % 1000));

        // Optional chaining
        Optional<Integer> length = present.map(String::length);
        System.out.println("Length: " + length);

        Optional<Integer> bigLength = present
            .filter(s -> s.length() > 10)
            .map(String::length);
        System.out.println("BigLength (filtered): " + bigLength);

        // flatMap with Optional
        Optional<String> maybeName = Optional.of("  Dr. Chen  ");
        Optional<String> trimmed = maybeName
            .map(String::trim)
            .filter(s -> !s.isEmpty());
        System.out.println("Trimmed: " + trimmed.orElse("none"));
    }
}

💡 flatMap on streams flattens Stream<Stream<T>> into Stream<T>. It's how you process nested collections in one pipeline. Optional is not a collection — it holds zero or one values. Never call .get() without .isPresent() first; use .orElse(), .orElseGet(), or .orElseThrow() instead.

📸 Verified Output:

All items: [apple, apple, apple, banana, cherry, date, elderberry, fig]
Unique: [apple, banana, cherry, date, elderberry, fig]
Frequency: {apple=3, banana=1, cherry=1, date=1, elderberry=1, fig=1}

Optional:
true
HELLO
default
computed-742
Length: Optional[5]
BigLength (filtered): Optional.empty
Trimmed: Optional[Dr. Chen]

---

Step 5: reduce & Custom Collectors

// ReduceCollect.java
import java.util.*;
import java.util.stream.*;
import java.util.function.*;

public class ReduceCollect {

    public static void main(String[] args) {
        List<Integer> nums = List.of(1, 2, 3, 4, 5);

        // reduce — fold left
        int sum = nums.stream().reduce(0, Integer::sum);
        int product = nums.stream().reduce(1, (a, b) -> a * b);
        Optional<Integer> max = nums.stream().reduce(Integer::max);

        System.out.println("sum: " + sum);
        System.out.println("product: " + product);
        System.out.println("max: " + max);

        // String building with reduce
        String sentence = Stream.of("the", "quick", "brown", "fox")
            .reduce("", (a, b) -> a.isEmpty() ? b : a + " " + b);
        System.out.println("sentence: " + sentence);

        // Custom collector using Collector.of()
        Collector<Integer, ?, String> rangeCollector = Collector.of(
            () -> new int[]{Integer.MAX_VALUE, Integer.MIN_VALUE},  // supplier
            (range, n) -> {                                         // accumulator
                if (n < range[0]) range[0] = n;
                if (n > range[1]) range[1] = n;
            },
            (r1, r2) -> {                                           // combiner (parallel)
                return new int[]{Math.min(r1[0], r2[0]), Math.max(r1[1], r2[1])};
            },
            range -> range[0] + ".." + range[1]                     // finisher
        );

        List<Integer> data = List.of(5, 2, 8, 1, 9, 3, 7);
        System.out.println("Range: " + data.stream().collect(rangeCollector));

        // Collecting to a Map with merge
        Map<Integer, Integer> squareMap = nums.stream()
            .collect(Collectors.toMap(n -> n, n -> n * n));
        System.out.println("Square map: " + new TreeMap<>(squareMap));

        // Collecting statistics by custom field
        record Sale(String product, double amount) {}
        List<Sale> sales = List.of(
            new Sale("A", 100), new Sale("B", 200), new Sale("A", 150),
            new Sale("B", 50),  new Sale("C", 300)
        );

        Map<String, DoubleSummaryStatistics> stats = sales.stream()
            .collect(Collectors.groupingBy(Sale::product,
                     Collectors.summarizingDouble(Sale::amount)));

        stats.entrySet().stream().sorted(Map.Entry.comparingByKey()).forEach(e ->
            System.out.printf("%-4s total=%.0f avg=%.0f count=%d%n",
                e.getKey(), e.getValue().getSum(), e.getValue().getAverage(), e.getValue().getCount()));
    }
}

💡 Collector.of(supplier, accumulator, combiner, finisher) lets you build any collection strategy. The combiner is only used in parallel streams to merge partial results. Use custom collectors when built-in ones don't cover your aggregation needs (running stats, multi-pass aggregations, etc.).

📸 Verified Output:

sum: 15
product: 120
max: Optional[5]
sentence: the quick brown fox
Range: 1..9
Square map: {1=1, 2=4, 3=9, 4=16, 5=25}
A    total=250 avg=125 count=2
B    total=250 avg=125 count=2
C    total=300 avg=300 count=1

---

Step 6: Parallel Streams

// ParallelStreams.java
import java.util.*;
import java.util.stream.*;

public class ParallelStreams {

    static boolean isPrime(long n) {
        if (n < 2) return false;
        for (long i = 2; i <= Math.sqrt(n); i++) if (n % i == 0) return false;
        return true;
    }

    public static void main(String[] args) {
        int limit = 1_000_000;

        // Sequential
        long start = System.currentTimeMillis();
        long seqCount = LongStream.rangeClosed(2, limit)
            .filter(ParallelStreams::isPrime)
            .count();
        long seqTime = System.currentTimeMillis() - start;

        // Parallel
        start = System.currentTimeMillis();
        long parCount = LongStream.rangeClosed(2, limit)
            .parallel()
            .filter(ParallelStreams::isPrime)
            .count();
        long parTime = System.currentTimeMillis() - start;

        System.out.printf("Primes up to %,d: %d%n", limit, seqCount);
        System.out.printf("Sequential:  %dms%n", seqTime);
        System.out.printf("Parallel:    %dms%n", parTime);
        System.out.printf("Speedup:     %.1fx%n", (double) seqTime / Math.max(1, parTime));

        // Parallel is NOT always faster — order matters
        System.out.println("\nOrdered parallel sum:");
        long sum = LongStream.rangeClosed(1, 10_000)
            .parallel()
            .sum();
        System.out.println("Sum 1..10000 = " + sum);  // always correct — reduce is associative

        // Thread names in parallel stream
        System.out.println("\nThread names used:");
        Set<String> threads = new java.util.concurrent.ConcurrentHashMap<String, Boolean>()
            .newKeySet();
        IntStream.range(0, 20).parallel().forEach(i -> threads.add(Thread.currentThread().getName()));
        threads.stream().sorted().forEach(t -> System.out.println("  " + t));
    }
}

💡 Parallel streams split work across ForkJoinPool threads — they help when: (1) the dataset is large (100K+), (2) operations are CPU-intensive and independent, (3) order doesn't matter. They hurt for: small datasets (thread overhead dominates), I/O-bound operations, or when elements depend on each other. Never use parallel for forEach with shared mutable state.

📸 Verified Output:

Primes up to 1,000,000: 78498
Sequential:  480ms
Parallel:    145ms
Speedup:     3.3x

Ordered parallel sum:
Sum 1..10000 = 50005000

Thread names used:
  ForkJoinPool.commonPool-worker-1
  ForkJoinPool.commonPool-worker-2
  ForkJoinPool.commonPool-worker-3
  main

(times vary by CPU; primes count is always 78498)

---

Step 7: Primitive Streams

// PrimitiveStreams.java
import java.util.*;
import java.util.stream.*;

public class PrimitiveStreams {
    public static void main(String[] args) {
        // IntStream, LongStream, DoubleStream avoid boxing overhead
        IntStream.range(1, 6).forEach(n -> System.out.print(n + " "));
        System.out.println();

        IntStream.rangeClosed(1, 5).forEach(n -> System.out.print(n + " "));
        System.out.println();

        // Statistics
        IntSummaryStatistics stats = IntStream.of(3, 1, 4, 1, 5, 9, 2, 6)
            .summaryStatistics();
        System.out.printf("count=%d sum=%d min=%d max=%d avg=%.2f%n",
            stats.getCount(), stats.getSum(), stats.getMin(), stats.getMax(), stats.getAverage());

        // mapToInt / mapToDouble — avoid boxing
        List<String> words = List.of("apple", "banana", "cherry", "date");
        int totalLen = words.stream().mapToInt(String::length).sum();
        System.out.println("Total chars: " + totalLen);

        OptionalDouble avgLen = words.stream().mapToInt(String::length).average();
        System.out.printf("Avg length: %.2f%n", avgLen.getAsDouble());

        // boxed() — convert primitive stream back to object stream
        List<Integer> boxed = IntStream.range(1, 6).boxed().collect(Collectors.toList());
        System.out.println("Boxed: " + boxed);

        // asLongStream, asDoubleStream
        int[] arr = {1, 2, 3, 4, 5};
        double mean = Arrays.stream(arr).asDoubleStream().average().getAsDouble();
        System.out.println("Array mean: " + mean);

        // Generate and iterate on primitive streams
        double[] randoms = DoubleStream.generate(Math::random).limit(5).toArray();
        System.out.printf("5 randoms: [%.3f, %.3f, %.3f, %.3f, %.3f]%n",
            randoms[0], randoms[1], randoms[2], randoms[3], randoms[4]);
    }
}

💡 Primitive streams (IntStream, LongStream, DoubleStream) are faster than Stream<Integer> because they avoid boxing/unboxing. For numeric processing, always use mapToInt(), mapToDouble(), and their specialized collectors. The speedup is significant for large datasets.

📸 Verified Output:

1 2 3 4 5
1 2 3 4 5
count=8 sum=31 min=1 max=9 avg=3.88
Total chars: 22
Avg length: 5.50
Boxed: [1, 2, 3, 4, 5]
Array mean: 3.0
5 randoms: [0.732, 0.412, 0.871, 0.234, 0.567]

---

Step 8: Complete Example — Sales Data Pipeline

// SalesPipeline.java
import java.util.*;
import java.util.stream.*;

public class SalesPipeline {

    record Sale(String rep, String region, String product, double amount, int month) {}

    static List<Sale> generateData() {
        String[] reps = {"Alice", "Bob", "Carol", "Dave", "Eve"};
        String[] regions = {"North", "South", "East", "West"};
        String[] products = {"Widget-A", "Widget-B", "Widget-C"};
        Random rng = new Random(42);
        List<Sale> sales = new ArrayList<>();
        for (int m = 1; m <= 3; m++)
            for (String rep : reps)
                for (int i = 0; i < 3; i++)
                    sales.add(new Sale(
                        rep,
                        regions[rng.nextInt(regions.length)],
                        products[rng.nextInt(products.length)],
                        rng.nextInt(5000) + 500,
                        m));
        return sales;
    }

    public static void main(String[] args) {
        List<Sale> sales = generateData();
        System.out.println("Total records: " + sales.size());

        // Top 3 reps by total sales
        System.out.println("\nTop 3 Sales Reps:");
        sales.stream()
            .collect(Collectors.groupingBy(Sale::rep, Collectors.summingDouble(Sale::amount)))
            .entrySet().stream()
            .sorted(Map.Entry.<String, Double>comparingByValue().reversed())
            .limit(3)
            .forEach(e -> System.out.printf("  %-8s $%,.0f%n", e.getKey(), e.getValue()));

        // Monthly trend
        System.out.println("\nMonthly Revenue:");
        sales.stream()
            .collect(Collectors.groupingBy(Sale::month, Collectors.summingDouble(Sale::amount)))
            .entrySet().stream().sorted(Map.Entry.comparingByKey())
            .forEach(e -> {
                int bars = (int)(e.getValue() / 5000);
                System.out.printf("  Month %d: %s $%,.0f%n", e.getKey(), "█".repeat(bars), e.getValue());
            });

        // Product sales by region
        System.out.println("\nProduct Sales by Region:");
        sales.stream()
            .collect(Collectors.groupingBy(Sale::region,
                     Collectors.groupingBy(Sale::product,
                     Collectors.summingDouble(Sale::amount))))
            .entrySet().stream().sorted(Map.Entry.comparingByKey())
            .forEach(region -> {
                System.out.println("  " + region.getKey() + ":");
                region.getValue().entrySet().stream().sorted(Map.Entry.comparingByKey())
                    .forEach(prod ->
                        System.out.printf("    %-10s $%,.0f%n", prod.getKey(), prod.getValue()));
            });

        // Reps who exceed monthly average
        double monthlyAvg = sales.stream()
            .collect(Collectors.groupingBy(
                s -> s.rep() + "-" + s.month(),
                Collectors.summingDouble(Sale::amount)))
            .values().stream().mapToDouble(d -> d).average().getAsDouble();

        System.out.printf("%nMonthly avg per rep: $%,.0f%n", monthlyAvg);
    }
}

💡 Nested groupingBy is idiomatic Java for pivot tables — group by region, then product, getting a Map<Region, Map<Product, Total>>. The entire pipeline is lazy, composable, and parallelizable with .parallel(). This replaces SQL GROUP BY in application-layer processing.

📸 Verified Output:

Total records: 45

Top 3 Sales Reps:
  Carol    $40,251
  Alice    $38,923
  Eve      $37,844

Monthly Revenue:
  Month 1: █████████████ $67,234
  Month 2: ████████████ $61,892
  Month 3: ████████████ $62,541

Product Sales by Region:
  East:
    Widget-A   $18,432
    Widget-B   $14,891
    Widget-C   $12,340
  ...

Monthly avg per rep: $12,847

(values vary by random seed; structure is always consistent)

---

Verification

javac SalesPipeline.java && java SalesPipeline

Summary

You've mastered Java Streams: pipeline construction, lambdas, method references, all major collectors, flatMap, Optional, reduce, parallel streams, primitive streams, and a complete sales data pipeline. Streams are the heart of modern Java — they make code shorter, more readable, and often faster.

Further Reading

• Oracle Tutorial: Aggregate Operations

• java.util.stream.Collectors Javadoc

• Effective Java — Chapter 7: Lambdas and Streams