[자바 인 액션] 6. 스트림 데이터 수집

스트림 최종 연산

---

최종적으로 중간연산을 통해 구성된 파이프라인을 연산하여 데이터 들을 소비하는 동작

 

[컬렉터]

Collector 인터페이스 구현은 스트림의 요소를 어떤 식으로 도출할지 지정

 

 

[고급 리듀싱 컬렉터]

Collectors 클래스에서 제공하는 여러 팩토리 메서드 기능이 있다.

1. 스트림 요소를 하나의 값으로 리듀스하고 요약 : 리스트에서 총합 등
2. 요소 그룹화 : 결과를 각각 그룹화
3. 요소 분할 :  Predicate를 사용한 분할을 사용한다.

 

갯수를 세는 컬렉터

long test = transactions.stream().collect(Collectors.counting());
long test2 = transactions.stream().count();

 

최댓값 검색

Optional<Dish> notCalorieDish = menu.stream().collect(Collectors.maxBy(Comparator.comparing(Dish::getCalories)));
Optional<Dish> ntest = menu.stream().max(Comparator.comparing(Dish::getCalories));

최댓값도 아래와 같은 2가지 방식이 있다.

하나는 Collectors.maxBy()를 사용한 방식, 다른 하나는 Stream.max()를 사용한 방식이다.

 

둘다 모든 요소를 리듀싱 작업합니다.

먼저 Stream.min의 경우 스트림의 전체 요소 중에서 최소/최대 값을 찾아야 하는경우에 사용합니다.

하지만 Collectors.minBy의 경우 요소를 그룹화 하거나 각 그룹에서 최소 최대 값을 찾아야 하는 경우에 사용한다.

 

 

요약 연산

int sum2 = menu.stream().collect(Collectors.summingInt(Dish::getCalories));

Double avg = menu.stream().collect(Collectors.averagingInt(Dish::getCalories));

 

IntSummaryStatistics menuIntSummaryStatistics = menu.stream().collect(Collectors.summarizingInt(Dish::getCalories));

summarizaingInt의 경우 count, sum, min, average, max의 값들을 한 번에 계산해 주는 메서드 이다.

 

• summarizingDoule, summarizingLong이 있다.
• 이와 관련한 메서드는 LongSummaryStatistics, DoubleSummaryStatistics가 있다.

 

문자열 연결

String shortMenu = menu.stream().map(Dish::getName).collect(Collectors.joining(","));

joining  팩토리 메서드를 이용하면 각 객체의 toString 메서드를 호출한다.

또한 합치기 위해 내부적으로 StringBuilder를 이용해서 문자열을 만든다.

 

 

Collectors.reducing

int totalCalories =  menu.stream().collect(Collectors.reducing(0, Dish::getCalories, (i,j)-> i+j));

위에서 언급했던 모든 동작은 이 reducing으로 구현이 가능합니다.

 

 

[그룹화]

데이터 집합을 하나 이상의 특성으로 분류해서 그룹화

Map<Dish.Type List<Dish>> dishesByType = menu.stream().collect(groupingBy(Dish::getType);

groupingBy를 통해 특정 값들로 분류할 수 있다.

 

Map<Dish.Type, List<Dish>> ct = menu.stream().filter(dish -> dish.getCalories() > 500).collect(Collectors.groupingBy(Dish::getType));

위와 같은 스트림이 있다하자 이렇게 되면 먼저 필터링이 되기 때문에 특정 키가 아예 사라져서 Map에는 존재하지 않는 경우도 생긴다. 따라서 이를 해결하기 위해서는 collect내부에 groupingBy를 할 때 조건을 걸어줘야 하는데

 

Map<Dish.Type , List<Dish>> caloricDishesByType = menu.stream()
        .collect(
                Collectors.groupingBy(Dish::getType, 
                        Collectors.filtering(dish-> dish.getCalories() > 500, Collectors.toList()))
        );

아래와 같이 변경할 수 있다.

 

만약 Dish리스트가 아니라 Map<Dish.Type, String>으로 만들고 싶다면

Map<Dish.Type, List<String>> ct2 = menu.stream().collect(Collectors.groupingBy(Dish::getType, Collectors.mapping(Dish::getName, Collectors.toList())));

조건에 Collectors.mapping(Dish::getName, Collectors.toList())를 통해 이름으로 변경후 리스트로 변경해 주는 역할을 한다.

 

Map<Dish.Type, Long> ct3 = menu.stream().collect(Collectors.groupingBy(Dish::getType, Collectors.counting()));

Map<Dish.Type, Optional<Dish>> ct4 = menu.stream().collect(Collectors.groupingBy(Dish::getType, Collectors.maxBy(Comparator.comparing(Dish::getCalories))));

이와같이 위에서 진행했던 나머지 조건들도 동일하게 적용이 가능하다.

 

 

[분할]

groupingBy가 스트림에 존재하는 것을 분류하는 작업이라 하면, 분할은 partitioningBy를 통해 조건문을 사용해서 True, False만을 분할한다 생각하면 된다.

 

Map<Boolean, List<Dish>> partitionMenu = menu.stream().collect(Collectors.partitioningBy(Dish::isVeg));

사용은 이전과 같지만 내부에는 Predicate를 사용한다 생각하면 된다.

 

List<Dish> vegetarianDishes = menu.stream().filter(Dish::isVeg).collect(Collectors.toList());

이전에는 위와같이 필터를 통해 둘 중 하나만을 선택했다고 하면

 

Map<Boolean, Map<Dish.Type, List<Dish>>> ct1 = menu.stream().collect(Collectors.partitioningBy(Dish::isVeg,
        Collectors.groupingBy(Dish::getType)));

이제는 해당 조건을 통해 True와 False로 구분이 가능해 진다.

 

이때 만약 Collectors 이후의 동작을 구현하고 싶다면, collectingAndThen와 같은 메서드를 참조하면 된다.

 

 

[Collector를 통한 최적의 컨테이너 만들기]

Collector인터페이스를 직접 구현함으로써 더 효율적으로 내가 가진 문제에 최적의 Collector를 만들 수 있다.

 

Collector 필수 메서드

• supplier : 새로운 결과 컨테이너 만들기
• accumulator : 결과 컨테이너에 요소 추가

• finisher :  최종 변화 값을 결과 컨테이너로 적용
• combiner : 두 결과 컨테이너 병합 (병렬 처리에서 사용)

 

동작 순서

1. supplier를 통한 결과 컨테이너 만들기
2. accumulator를 통한 컨테이너에 요소 반복 추가
3. finisher로 결과 값 반환

 

병렬시 동작 순서

1. 스트림을 분할하는 조건을 기준으로 조건 만족하지 않을 때까지 분할
   • 대부분 프로세싱 코어의 개수를 초과하지 않는다.
   • 초과하게 되면 오히려 속도가 느려진다 (각 코어마다 context 변경 비용이 있기 때문에)
2. 각 스트림은 supplier와 accumulator를 통해 요소를 추가한다.
3. 분할된 가장 스트림은 각 스트림끼리 combiner를 하는데 한쪽에 다른 한쪽을 전부 넣어주게 된다.
4. 독립적으로 처리된 결과를 마지막으로 finisher를 통해 결과를 반환하게 된다.

class ToListCollector<T> implements Collector<T, List<T>, List<T>> {

    @Override
    public Supplier<List<T>> supplier() {
        return ArrayList::new;
    }

    @Override
    public BiConsumer<List<T>, T> accumulator() {
        return List::add;
    }

    @Override
    public BinaryOperator<List<T>> combiner() {
        return (list1, list2) -> {
            list1.addAll(list2);
            return list1;
        };
    }

    @Override
    public Function<List<T>, List<T>> finisher() {
        return Function.identity();
    }

    @Override
    public Set<Characteristics> characteristics() {
        return Collections.unmodifiableSet(EnumSet.of(Characteristics.IDENTITY_FINISH, Characteristics.CONCURRENT));
    }
}

위는 각 동작에 대한 각각의 과정이다.

 

Characteristics 메서드

• 컬렉터의 연산을 정의하는 형식
• UNORDERED: 리듀싱 결과는 스트림 요소의 방문 순서나 누적 순서에 영향을 받지 않는다.
• CONCURRENT: 다중 스레드에서 accumulator를 동시에 호출하며 스트림 병렬 리듀싱을 수행한다.
• IDENTITY_FINISH: identity를 적용할 뿐 생략 가능하다. 

만약 내가 작업하는 것이 순서에 상관이 없고, 추가 작업이 없고 데이터의 양이 많아 병렬로 처리한다.

UNORDERED & IDENTITY_FINISH & CONCURRENT이다.

 

하지만, 데이터 양이 많아도 작업하는데 순서에 상관이 있다면

IDENTITY_FINISH만 가능하다. 순서로 인해 병렬처리를 하기 힘들기 때문이다.