[모던 인 자바 액션] 스트림으로 데이터 수집

이주오·2021년 8월 23일
Java
0

도서

목록 보기
6/15

예시로 시작해보자!!

  • 통화별로 트랜잭션을 그룹화한 다음에 모든 트랜잭션 합계를 계산하는 예제
  • 즉 Map<Currency, Integer> 을 반환해야 한다.
  • 명령형 코드
    • 코드가 길어서 무엇을 실행하는지 파악하기 어렵다.
    Map<Currency, List<Transaction>> transactionByCurrencies = new HashMap<>(); //  그룹화한 트랜잭션을 저장할 맵을 생성

    for (Transactin transaction : transactions) {   //  트랜잭션 리스트를 반복
        Currency currency = transaction.getCurrency();
        List<Transaction> transactionsForCurrency = transactionsByCurrencies.get(currency);

        if (transactionForCurrency = null) {    //  현재 통화를 그룹화하는 맵에 항목이 없으면 항목을 만든다.
            transactionsForCurrency = new ArrayList<>();
            transactionsByCurrencies.put(currency, transactionsForCurrency);
        }

        transactinsForCurrency.add(transaction);    //  같은 통화를 가진 트랜잭션 리스트에 현재 탐색 중인 트랜잭션을 추가
    }
  • 함수형 코드
    Map<Currency, List<Transaction>> transactionsByCurrencies = transactions.stream().collect(groupingBy(Transaction::getCurrency));

이번 주제 키워드

reduce 연산(하나로 수렴)을 통해 스트림을 다양한 형태로 collect 하는 방법에 대한 내용

  • Collectors
  • reduce, reducing
  • 데이터의 그룹화(groupingBy)와 분할(Predicate)
  • Collector 인터페이스 살펴보기
    • toList() 살펴보기

Collector란??

  • 자바 8에서 Collector는 Stream 요소들을 어떻게 분류하고 모을지(reducing)를 담당하는 인터페이스이다.
    • 즉 고급 리듀싱 기능을 수행
  • Collector 인터페이스를 implement하고 구현하여 스트림의 요소를 어떤 식으로 도출할지 지정한다.
  • 스트림에 collect()를 호출하면 스트림의 요소에 (컬렉터로 파라미터화된) 리듀싱 연산이 수행된다.
    • 즉 collect에서는 리듀싱 연산을 이용해서 스트림의 각 요소를 방문하면서 컬렉터가 작업을 처리한다.
    • 따라서 Collector 인터페이스의 메서드를 어떻게 구현하느냐에 따라 스트림에 어떤 리듀싱 연산을 수행할지 결정


미리 정의된 컬렉터

Collectors 유틸티리 클래스에는 자주 사용하는 컬렉터 인스턴스를 쉽게 생설할 수 있는 정적 팩터리 메서드를 제공한다.

Collectors에서 제공하는 메서드의 기능은 크게 세가지

  1. 스트림 요소를 하나의 값으로 reduce하고 요약
  2. 스트림 요소들의 그룹화
  3. 스트림 요소들의 분할 (이것도 결국 bool 그룹화)

첫번째, Reducing과 요약

다양한 계산을 수행할 때 사용하는 유용한 컬렉터

  • 스트림에 있는 객체의 숫자 필드의 합계나 평균 등을 반환하는 연산

counting()

  • 다른 컬렉터와 함께 사용할 때 유용하다.
menu.stream().collect(Collectors.counting())
menu.stream().count()

maxBy(), minBy()

  • 스트림값에서 최댓값과 최솟값 검색할 때 사용하는 컬렉터
  • 해당 컬렉터는 스트림의 요소를 비교하는데 사용할 Comparator인수를 받는다
Optional<Dish> mostCalorieDish = menu.stream().collect(Collectors.maxBy(Comparator.comparingInt(Dish::getCalories)));

요약 연산

  • 필드의 합계나 평균 등을 반환하는 연산
  • 리듀싱 기능이 자주 사용된다.
  • 대표적으로 Collectors.summingInt
public static <T> Collector<T, ?, Integer> summingInt(ToIntFunction<? super T> mapper) {
  return new CollectorImpl<>(
          () -> new int[1],
          (a, t) -> { a[0] += mapper.applyAsInt(t); },
          (a, b) -> { a[0] += b[0]; return a; },
          a -> a[0], CH_NOID);
}
@FunctionalInterface
public interface ToIntFunction<T> {

    /**
     * Applies this function to the given argument.
     *
     * @param value the function argument
     * @return the function result
     */
    int applyAsInt(T value);
}
int totalColories = menu.stream().collect(summingInt(Dish:getCalories));
  • 위의 코드처럼 객체를 int로 매핑하는 함수를 인수로 받는다.
  • 인수로 전달된 함수는 객체를 int로 매핑한 컬렉터를 반환한다.
  • summingInt가 collect 메서드로 전달되면 요약 작업을 수행한다.
  • 이외에도 averagingInt 메서드도 존재한다.

만약 두 개 이상의 연산을 한번에 수행해야 할 때는?

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

문자열 연결

  • joining()
  • 스트림의 각 객체에 toString 메서드를 호출해서 추출한 모든 문자열을 하나의 문자열로 연결해서 반환한다.
menu.stream().map(Dish::getName).collect(Collectors.joining(", "));

범용 리듀싱 요약 연산

이러한 모든 Collector는 reducing 팩토리 메서드로도 정의할 수 있다. (가독성이나 편리성 측면에서 권장하지 않는다)

  • reducing()
public static <T, U> Collector<T, ?, U> reducing(
		U identity,
		Function<? super T, ? extends U> mapper,
		BinaryOperator<U> op) {
	...
}
  • 첫 번째 인수 : reducing 연산의 초기값. 스트림에 인수가 없을 때는 반환값.
  • 두 번째 인수 : 요소를 변환하는 변환함수.
  • 세 번째 인수 : 같은 종류의 두 항목을 하나의 값으로 더하는 합계 함수(BinaryOperator.)
  • 예시 - 모든 메뉴 요소의 칼로리 합계
    menu.stream().collect(Collectors.reducing(0, Dish::getCalories, (i, j) -> i + j)
    menu.stream().collect(Collectors.reducing(0, Dish::getCalories, Integer::sum)

한개짜리 인수를 갖는 reducing()

public static <T> Collector<T, ?, Optional<T>> reducing(BinaryOperator<T> op) {
	...
}
  • 초기값은 스트림의 첫 요소
  • 두번째 인수는 자신을 그대로 반환하는 항등 함수
  • 따라서 빈 스트림인 경우를 대비해 Optional를 반환한다.

잠깐 BinaryOperator을 살펴보자

public interface BinaryOperator<T> extends BiFunction<T,T,T> {
}

@FunctionalInterface
public interface BiFunction<T, U, R> {
    R apply(T t, U u);
}
  • 즉 두 인수의 타입과 반환하는 타입은 같아야 한다.
    • 따라서 아래와 같은 사용은 컴파일 error
menu.stream().collect(Collectors.reducing((d1, d2) -> d1.getName() + d2.getName()))

collect vs reduce

  • 이 두가지 메서드로 같은 기능을 구현할 수 있지만 의미론적인 문제와 실용성 문제 등 몇가지 문제가 있다.
  • 의미론적인 문제
    • collect 메서드는 도출하려는 결과를 누적하는 컨테이너를 바꾸도록 설계된 메서드
    • reduce 메서드는 두 값을 하나로 도출하는 불변형 연산
  • 실용성 문제
    • 여러 스레드가 동시에 같은 데이터 구조체를 고치면 리스트 자체가 망가져버리므로 리듀싱 연산을 병렬로 수행할 수 없다.
    • 이를 피하기 위해 매번 리스트를 새로 할당하고 객체를 할당하느라 성능이 낮을 것 이다.

두번째, 그룹화

맨 처음 트랜잭션 currency 그룹화 예제를 생각해보면 명령형으로 그룹화를 구현하려면 까다롭고, 에러도 신경써야 하고, 코드가 길다.

  • Collectors.groupingBy 이용하면 쉽게 그룹화가 가능하다.
  • gropingBy 메서드의 인수로 전달되는 함수를 기준으로 그룹화되므로 이를 분류 함수라고 부른다.

1. 타입에 따른 메뉴 그룹 구하기

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

2. 복잡한 기준으로 그룹화하기

  • 메서드 참조 대신 람다 사용
public enum CaloricLevel { DIET, NORMAL, FAT }

Map<CaloricLevel, List<Dish>> dishesByCaloricLevel = menu.stream().collect(
	groupingBy(dish -> {
    if (dish.getCalories() <= 400) {
        return CaloricLevel.DIET;
    } else if (dish.getCalories() <= 700) {
        return CaloricLevel.NORMAL;
    } else {
        return CaloricLevel.FAT;
    }
	}));

3. 다수준 그룹화

Map<Dish.Type, Map<CaloricLevel, List<Dish>>> dishesByTypeCaloricLevel = 
    menu.stream().collect(
        groupingBy(Dish::getType,
            groupingBy(dish -> {
                if (dish.getCalories() <= 400) {
                    return CaloricLevel.DIET;
                } else if (dish.getCalories() <= 700) {
                    return CaloricLevel.NORMAL;
                } else {
                    return CaloricLevel.FAT;
                }
            })
        )
    );

// 결과
{FISH={NORMAL=[salmon], DIET=[prawns]}, OTHER={NORMAL=[fries, pizza], DIET=[rice, fruit]}, MEAT={FAT=[pork], NORMAL=[beef], DIET=[chicken]}}
  • n수준 그룹화를 통해 n 수준 맵이 된다.

4. 서브그룹으로 데이터 수집

  • 분류 함수 한 개의 인수를 갖는 groupingBy(f)는 사실 groupingBy(f, toList())의 축약형이다.
  • 따라서 요리의 종류를 분류하는 컬렉터로 메뉴에서 가장 높은 칼로리를 가진 요리를 찾는 로직도 구현할 수 있다.
Map<Dish.Type, Optional<Dish>> mostCaloricByType = 
    menu.stream().collect(groupingBy(Dish::getType, maxBy(comparingInt(Dish::getCalories))));

// 결과
{OTHER=Optional[pizza], MEAT=Optional[pork], FISH=Optional[salmon]}

5. 컬렉션 결과를 다른 형식에 적용하기

  • collectingAndThen은 적용할 컬렉터와 변환 함수를 인수로 받아 다른 컬렉터를 반환한다.
  • 반환되는 컬렉터는 기존 컬렉터의 래퍼 역할을 하며 collect 마지막 과정에서 변환 함수로 자신이 반환하는 값을 매핑한다.
Map<Dish.Type, Dish> mostCaloricByType =
    menu.stream()
        .collect(groupingBy(Dish::getType,  //  분류함수
                 collectingAndThen(maxBy(comparingInt(Dish::getCalories)),  //  감싸인 컬렉터
                 Optional::get)));  //  변환함수. Optional에 포함된 값을 추출

세번째, 분할

분할은 분할 함수(프레디케이트)를 분류 함수로 사용하는 특수한 그룹화다

  • 프레디케이트를 사용하므로 맵의 키 형식은 bool
  • 따라서 두 개의 그룹으로 분류된다.

채식요리와 아닌 요리 분류

Map<Boolean, List<Dish>> partitionedMenu = menu.stream().collect(partitioningBy(Dish::isVegetarian));

// 결과 : {false=[pork, beef, chicken, prawns, salmon], true=[french fries, rice, season fruit, pizza]}

분할의 장점

  • filter를 사용해서도 간단히 원하는 내용을 얻을 수 있는데 왜 사용해야하는가?
  • 참, 거짓 두가지 요소의 스트림 리스트를 모두 유지한다는 것이 장점이다.
    • filter는 말그대로 필터링하여 보여주는 것이고, 분할은 데이터를 분할하는 용도로 사용하는 것이다.
  • 두 번째 인수로 컬렉터를 사용하는 오버로드 메서드도 있다.
Map<Boolean, Map<Dish.Type, List<Dish>>> vegetarianDishesByType = menu.stream().collect(
	partitioningBy(Dish::isVegetarian,  //  분할 함수
		groupingBy(Dish::getType)));    //  두 번째 컬렉터

// 결과 : 
{false={MEAT=[pork, beef, chicken], FISH=[prawns, salmon]}, 
 true={OTHER=[french fries, rice, season fruit, pizza]}}

지금까지 살펴본 모든 Collector는 Collector 인터페이스를 구현한다. Collector 인터페이스를 자세히 살펴보자.


Collector 인터페이스란?

  • reducing 연산(Collector)를 어떻게 구현할지 제공하는 메서드 집합으로 구성된 인터페이스.
  • Collector 인터페이스를 직접 구현하면 문제를 더 효율적으로 해결할 수 있다.
public interface Collector<T, A, R> {
    Supplier<A> supplier();
    BiConsumer<A, T> accumulator();
    BinaryOperator<A> combiner();
    Function<A, R> finisher();
    Set<Characteristics> characteristics();
  • T는 수집된 스트림 항목의 제네릭 타입
  • A는 누적자, 즉 수집 과정에서 중간 결과를 누적하는 객체의 타입
  • R은 수집 연산 결과 객체의 타입
    • 주로 컬렉션 타입

toList()를 확인해보면서 파악해보자!!

public static <T> Collector<T, ?, List<T>> toList() {
    return new CollectorImpl<>((Supplier<List<T>>) ArrayList::new, List::add,
                               (left, right) -> { left.addAll(right); return left; },
                               CH_ID);
}

supplier 메서드

새로운 결과 컨테이너 만들기

  • supplier 메서드는 빈 결 이루어진 Supplier를 반환해야 한다.
@FunctionalInterface
public interface Supplier<T> {
    T get();
}
  • 즉, supplier는 수집 과정에서 빈 누적자 인스턴스를 만드는 파라미터가 없는 함수다.
  • toList()에서 supplier()은 빈 리스트를 반환한다.
public Supplier<List<T>> supplier() {
	return () -> new ArrayList<T>();
}

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

accumulator 메서드

결과 컨테이너에 요소 추가하기

@FunctionalInterface
public interface BiConsumer<T, U> {
    void accept(T t, U u);
  • 스트림에서 n번째 요소를 탐색할 때 누적자와 n번째 요소를 함수에 적용한다.,
  • 함수의 반환값은 void, 즉 요소를 탐색하면서 적용하는 함수에 의해 누적자 내부 상태가 바뀌므로 누적자가 어떤 값일지 단정할 수 없다.
  • toList()
public BiConsumer<List<T>, T> accumulator() {
    return (list, item) -> list.add(item);
}

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

finisher 메서드

최종 변환 값을 결과 컨테이너로 적용하기

  • finisher 메서드는 스트림 탐색을 끝내고 누적자 객체를 최종 결과로 변환하면서 누적 과정을 끝낼 때 호출할 함수(Function<T, R>)를 반환해야 한다.
@FunctionalInterface
public interface Function<T, R> {
    R apply(T t);

		static <T> Function<T, T> identity() {
		        return t -> t;
		}
}
  • 때로 누적자 객체가 이미 최종 결과인 상황도 있는데 이럴 때는 변환 과정이 필요하지 않으므로 finisher 메서드는 항등 함수를 반환한다.
    • toList()도 마찬가지
public Function<List<T>, List<T>> finisher() {
    return Function.identity();
}

위 3가지 메소드는 순차적 스트림 리듀싱 기능을 수행할 수 있다. 하지만 실제로 collect가 동작하기 전, 다른 중간 연산과 파이프라인을 구성할 수 있게 해주는 게으른 특성 그리고 병렬 실행 등도 고려해야 하므로 스트림 리듀싱 기능 구현은 생각보다 복잡하다.

combiner 메서드 : 두 결과 컨테이너 병합

  • combiner는 리듀싱 연산에서 사용할 함수를 반환하는 메서드이다.
  • combiner는 스트림의 서로 다른 서브파트를 병렬로 처리할 때 누적자가 이 결과를 어떻게 처리할 지 정의한다.
public interface BinaryOperator<T> extends BiFunction<T,T,T> {
}

@FunctionalInterface
public interface BiFunction<T, U, R> {
    R apply(T t, U u);
}
  • toList()의 경우
    • 스트림의 두번째 서브 파트에서 수집한 항복 리스트를 첫 번째 서브파트 결과 리스트의 뒤에 추가하면 된다.
public BinaryOperator<List<T>> combiner() {
    return (list1, list2) -> {
        list1.addAll(list2);
        return list1;
    }
}
  • combiner 메서드를 이용하면 스트림의 리듀싱을 병렬로 수행할 수 있다.
  • 이 때 자바7의 포크/조인 프레임워크와 Spliterator를 사용한다.

  1. 스트림의 분할을 정의하는 조건이 성립할 경우 원래 스트림을 재귀적으로 분할한다.
  2. 서브스트림의 각 요소에 리듀싱 연산을 순차적으로 적용하여 서브스트림을 병렬로 처리한다.
  3. combiner 메서드가 반환하는 함수를 통해 각 서브스트림의 결과를 합쳐 연산을 완료한다.

Characteristics 메서드

  • Characteristics 메서드는 컬렉터의 연산을 정의하는 Charactieristics 형식의 불변 집합을 반환한다.
  • Characteristics는 스트림을 병렬로 리듀스할건지, 그리고 병렬로 리듀스한다면 어떤 최적화를 선택해야 할지 힌트를 제공한다.
  • Characteristics는 다음 세 항목의 특성을 갖는 열거형이다.

UNORDERED

  • 리듀싱 결과는 스트림의 방문 순서나 누적 순서에 영향을 받지 않는다.

CONCURRENT

  • 다중 스레드에서 accumulator 함수를 동시에 호출할 수 있으며 이 컬렉터는 스트림의 병렬 리듀싱을 수행할 수 있다.
  • 컬렉터의 플래그에 UNORDERED를 함께 설정하지 않았다면 데이터 소스가 정렬되어 있지 않은(집합처럼 요소의 순서가 무의미한) 상황에서만 병렬 리듀싱을 수행할 수 있다.

DENTITY_FINISH

  • finisher 메서드가 반환하는 함수는 identity를 적용할 뿐이므로 이를 생략할 수 있다.

  • 따라서 리듀싱 과정의 최종 결과로 누적자 객체를 바로 사용할 수 있다.

  • 또한 누적자 A를 결과 R로 안전하게 형변환 할 수 있다.

  • toList()

    • 마찬가지로 스트림의 요소를 누적하는데 사용한 리스트가 최종 결과 형식이므로 추가 변환이 필요 없다.
    static final Set<Collector.Characteristics> CH_ID
                = Collections.unmodifiableSet(EnumSet.of(Collector.Characteristics.IDENTITY_FINISH));

요약

  • collect는 스트림의 요소를 요약 결과로 누적하는 다양한 방법(컬렉터)을 인수로 갖는 최종 연산이다.
  • 다양한 컬렉터 등이 Collections 유틸 클래스에 미리 정의되어 있다.
  • 컬렉터는 다수준의 그룹화, 분할, 리듀싱 연산에 적합하게 설계되어 있다.
  • Collector 인터페이스에 정의된 메서드를 구현해서 커스텀 컬렉터를 개발할 수 있다,

참고 출처

profile
이주오
동료들이 같이 일하고 싶어하는 백엔드 개발자가 되고자 합니다!
이전 포스트

[모던 자바 인 액션] 스트림 활용

다음 포스트

[모던 인 자바 액션] 병렬 데이터 처리와 성능

0개의 댓글