Stream MapReduce and Aggregation

import java.util.*;
import java.util.stream.*;

public class MapReduceDemo {
    record Sale(String product, String region, double amount, int qty) {}

    public static void main(String[] args) {
        var sales = List.of(
            new Sale("Widget", "US", 29.99, 10),
            new Sale("Gadget", "US", 49.99, 5),
            new Sale("Widget", "EU", 34.99, 8),
            new Sale("Widget", "US", 29.99, 15),
            new Sale("Gadget", "EU", 54.99, 3),
            new Sale("Gizmo", "US", 19.99, 20),
            new Sale("Gizmo", "EU", 24.99, 12)
        );

        // 1. Total revenue
        double totalRevenue = sales.stream()
            .mapToDouble(s -> s.amount() * s.qty())
            .sum();
        System.out.printf("Total Revenue: $%,.2f%n", totalRevenue);

        // 2. Revenue by product
        Map<String, Double> byProduct = sales.stream()
            .collect(Collectors.groupingBy(
                Sale::product,
                Collectors.summingDouble(s -> s.amount() * s.qty())
            ));
        System.out.println("\nBy Product:");
        byProduct.forEach((p, rev) -> System.out.printf("  %s: $%,.2f%n", p, rev));

        // 3. Revenue by region and product (pivot table)
        Map<String, Map<String, Double>> pivot = sales.stream()
            .collect(Collectors.groupingBy(
                Sale::region,
                Collectors.groupingBy(
                    Sale::product,
                    Collectors.summingDouble(s -> s.amount() * s.qty())
                )
            ));
        System.out.println("\nPivot:");
        pivot.forEach((region, products) -> {
            System.out.println("  " + region + ":");
            products.forEach((p, rev) -> System.out.printf("    %s: $%,.2f%n", p, rev));
        });

        // 4. Top product by revenue
        String topProduct = sales.stream()
            .collect(Collectors.groupingBy(
                Sale::product,
                Collectors.summingDouble(s -> s.amount() * s.qty())
            ))
            .entrySet().stream()
            .max(Map.Entry.comparingByValue())
            .map(Map.Entry::getKey)
            .orElse("None");
        System.out.println("\nTop Product: " + topProduct);

        // 5. Statistics per product
        Map<String, DoubleSummaryStatistics> stats = sales.stream()
            .collect(Collectors.groupingBy(
                Sale::product,
                Collectors.summarizingDouble(Sale::amount)
            ));
        stats.forEach((p, s) ->
            System.out.printf("%s: avg=$%.2f, min=$%.2f, max=$%.2f, count=%d%n",
                p, s.getAverage(), s.getMin(), s.getMax(), s.getCount()));

        // 6. Custom reduce — build report string
        String report = sales.stream()
            .reduce(
                new StringBuilder(),
                (sb, s) -> sb.append(String.format("%s/%s: $%.2f x %d%n",
                    s.region(), s.product(), s.amount(), s.qty())),
                StringBuilder::append
            ).toString();
        System.out.println("\nReport:\n" + report);
    }
}