Java 8 Steam 관련 자료

이번에는 Java SE 8에서 추가된 아주 좋은 SteamAPI에 대해서 알아 보도록 하겠습니다.

오역이 있을 수도 있으니 원본 문서도 한번 보는것을 추천 드립니다. 

원본 글  : http://www.oracle.com/technetwork/articles/java/ma14-java-se-8-streams-2177646.html

오타 및 오역 지적해주시면 감사하겠습니다. 

============================================================================================

복잡한 데이터 처리 질의를 표현하기 위해 Stream을 사용!.

대부분의 자바 애플리케이션들이 데이터를 만들고 처리하기 위해서 컬렉션을 사용한다. 이 컬렉션을 처리하는 로직 또한 같이 구현되어 코드에 포함되어 있고 이는 많은 프로그램의 기본(데이터를 그룹핑 하고 제어 하는 등)이 된다. 

예를 들어 은행에서 고객들의 입출력 내역을 가지고 있는 컬렉션을 생성하고 이중에서 고객이 사용한 내역들에 대해 추출하는 로직을 작성할 수 있을 것이다. 이런 은행 관련 중요한 기능들을 컬렉션을 통해 작정할 수 있지만 코드가 그다지 아름답지 않고 지저분하게 보여진다. 

컬렉션을 사용하는데는 두가지 단점이 있다.

1. 컬렉션에서 데이터를 처리하는 전형적인 처리 패턴은 SQL에서 데이터를 찾는것과 비슷하거나(처리 내역 중 가장 높은 값을 찾기) 데이터를 묶는것(구매 내역중에 식료품에서 쇼핑한 내역만 찾기)과 비슷하다. 대부분의 데이터베이스들은 이를 명확하게 지원하는데 만약 가장 높은 값을 가지는 것을 찾고 싶으면 "SELECT ID, MAX(VALUE) FROM TRANSACTIONS"를 사용 하면 된다.

위의 예에서 보는것 처럼 최대값을 어떻게 계산해야 하는지에 대해서 직접 구현할 필요가 없이(반복문을 사용해서 가장 높은 값을 찾는 것) 단지 원하는 값을 표현해 주기만 하면 된다. 컬렉션도 데이터베이스처럼 데이터의 집합인데 왜 비슷한 방법으로 처리할 생각을 못하고 이와 비슷한 구현을 하기 위해서 얼마나 많은 시간을 소모해서 복잡한 반복문을 작성하고 또 작성했는지 생각해보자.

2. 만약 큰 사이즈의 컬렉션은 어떻게 처리 해야 할까? 처리 속도를 향상하기 위해서는 병렬 처리를 하도록 코드를 작성해야 되는데 이는 기술적으로도 어렵고 많은 에러가 발생한다.

이러한 단점들은 Java SE 8이 출시되면서 해결할 수 있게 되었다. Stream 이라는 새로운 API가 공개되었고 이를 통해 쿼리를 작성하듯 데이터를 명시적인 방법으로 처리할 수 있게 되었다. 게다가 Stream은 멀티 스레드 관련 코드도 별도로 작성할 필요 없이 멀티코어를 지원할 수 있게 되었다. 좋아 보이지 않는가? 이 문서에서는 이런 내용들에 대하여 살펴볼 것이다.

Stream을 가지고 무엇을 할 수 있는지 알아보기 전에 Java SE 8의 Stream 사용법에 대해 간단히 살펴보자. 식료품점에서 쇼핑한 모든 거래내역중에 거래 ID를  가장 큰 비용이 들어간 순으로 정렬해서 추출하는 것에 대한 코드를 작성해보자.

Listing 1. Java SE 7 이하에서의 처리 방법

List groceryTransactions = new ArrayList();
for (Transaction t : groceryTransactions) {
    if (t.getType() == Transaction.GROCERY) {
        groceryTransactions.add(t);
    }
}
Collections.sort(groceryTransactions, new Comparator() {
    public int compare(Transaction t1, Transaction t2) {
        return t2.getValue().compareTo(t1.getValue());
    }
});

List transactionIds = new ArrayList();
for (Transaction t : groceryTransactions) {
    transactionIds.add(t.getId());
}

Listing 2. Java SE 8 에서의 처리 방법

List transactionsIds = 
    transactions.stream()
        .filter(t -> t.getType() == Transaction.GROCERY)
        .sorted(Comparator.comparing(Transaction::getValue).reversed())
        .map(Transaction::getId)
        .collect(Collectors.toList());

아래 그림 Fiqure 1은 Java SE 8에서 Stream이 동작하는 과정을 그린것이다.

먼저 모든 거래내역(List)로 부터 stream()메서드를 사용해서 stream을 가져오고 그 다음에 여러가지 기능(filter, sorted, map,collect)를 체인 처럼 엮어서 데이터를 처리하는 쿼리처럼 보이게 만든다.

Figure 1

병렬 처리 코드 작성은 Java SE 8에서 매우 쉽게할 수 있다. Listing 3. 처럼 그냥 stream()을 parallelStream()으로 변경만 하면 된다. Stream API는 내부적으로 멀티코어로 동작하도록 처리한다.

Listing 3. 

List transactionsIds = 
    transactions.parallelStream()
        .filter(t -> t.getType() == Transaction.GROCERY)
        .sorted(Comparator.comparing(Transaction::getValue).reversed())
        .map(Transaction::getId)
        .collect(Collectors.toList());

처음보는 형태의 코드를 보고 걱정할 필요는 없다. 다음 섹션에서 이것들이 어떻게 동작하는지 살펴볼 것이다. 하지만 람다 표현식(t-> t.gerCategory() == Transaction.GROCERY)의 사용방법이나 메소드 참조(method references, Transaction::getId)에 대해서는 미리 한번 보는것이 좋다.

지금 부터 컬렉션에 저장된 데이터들을 SQL이 동작하는 것 처럼 스트림을 활용해서 처리 하는것에 대해서 살펴볼 것이다. 이 모든 동작들은 람다 표현식과 함께 간단한 파라미터로 처리 될 수 있다. 

Java SE 8의 Stream에 대한 이 문서를 다 읽어보면 Stream API를 사용하여 위 예제와 같이 멋진 방식으로 사용할 수 있을 것이다.

Stream 시작 하기

작은 부분부터 시작하도록 해보자. Stream의 정의는 무엇일까? 간단하게 정의해 보면 "집계 연산을 지원하는 요소의 순서(a sequence of elements from a source that supports aggregate operations.)"라고도 할수 있는데 이에 대해 더 알아 보도록 하자.

• Sequence of element : Stream은 정의된 엘리먼트의 속성에 따라서 처리할 수 있는 인터페이스를 제공하지만 실제 엘리먼트들을 저장하지 않고 계산하는데만 쓰인다.
• Source : 스트림은 컬렉션, 배열, I/O 리소스 등에서 제공받은 데이터를 가지고 작업을 처리 한다.
• Aggreate operations : Stream은 SQL 같은 처리를 지원하고 함수형 프로그래밍 같은 처리 방법도 지원한다. (filter, map,reduce, find, match, sorted 등)

게다가 Stream은 기존의 컬렉션 처리 방법과 구분되게 하는 기본적인 두가지 특징이 있다.

• Pipelining : 많은 Stream 기능들이 Stream 자기 자신을 리턴한다. 이 방식은 처리 작업이 체인처럼 연결되어 큰 파이프라인처럼 동작 하도록 한다. 이를 통해 laziness와 short-circuiting 과 같이 최적화 작업을 할 수 있다.
• Internal iteration : 명시적으로 반복작업을 수행해야 되는 Collection과 비교 하면 Stream 작업은 내부에서 처리된다.

아래 그림(Figure 2) 은 Listing 2에 작성되어 있는 코드 상세한 그림으로 설명한 것이다.

Figure 2

처음 접한 Stream은 list에서 거래내역에 대한 목록을 stream() 메서드를 사용해서 Stream 으로 가져온 것이다. 그 다음은 몇가지 집계 관련된 작업을 진행하는데, filter(주어진 속성에 맞게 필터링), sorted(주어진 compator를 자기고 정렬), map(정보를 추출) 메서드를 사용해서 집계 관련된 처리를 수행한다. collect()를 제외한 모든 메서드들은 stream을 리턴하기 때문에 이들을 서로 연결해서 파이프라인형태로 처리할 수 있다.

이 작업들은 collect()가 호출되기전까지 실제로 실행되지 않는다. collect()는 파이프라인에 연결된 집계 작업을 시작하고 완료되면 결과를 리턴한다(Stream 형태가 아닌, 여기서는 list). collect()에 대해서는 나중에 좀 더 자세히 살펴볼테니 지금 당장 걱정할 필요는 없다. collect() 메서드를 보면 인자값을 가지고 있는것을 볼 수 있는데 이 파라미터를 가지고 리턴 값을 결정 한다. toLIst()는 Stream을 List로 변경하여 리턴해준다.

Stream에서 사용가능한 다른 메서드들에 대해 알아보기 전에 Stream과 Collection간의 개념적으로 다른점을 확인해 보도록 하자.

Stream VS Collections

예전부터 사용했던 Collection 과 이번에 새로 추가된 Stream은 엘리먼트들을 순차적으로 처리하는것과 관련된 인터페이스를 제공한다. 차이점을 간단히 말하면 Collection은 데이터와 관련된 것이고 Stream은 데이터를 계산하는것과 관련된 내용이다.

DVD에 저장된 영화를 예를 들어 생각해보자. 이것들은 데이터들로 구성되어 있기 때문에 Collection으로 볼 수 있다(바이트나 프레임으로 구성된 집합) 같은 영화를 인터넷 스트리밍을 통해서 본다고 가정해보자. 효율적으로 사용자들에게 비디오를 보여주기 위해 사용자가 보고 있는 부분들에 대해 미리 다운로드를 하는게 효율적이다. 이렇게 다운로드를 할 범위를 결정하는 것을 Stream으로 볼 수 있다.

간단하게 말하면 Collection과 Stream의 차이점은 무언가를 계산할때 라고 볼 수 있다 Collection은 메모리에 저장되는 데이터 구조고 Stream은 이를 계산해서 새로운 데이터를 만들어 낼 수 있다.

Collection 인터페이스를 사용하기 위해서는 사용자에 의해 정의된 반복문이 필요한데(for loop, foreach 등) 이를 외부 반복이라고 부른다.

이와 다르게 Stream은 내부 반복을 사용한다(내부적으로 반복작업을 수행하고 결과값을 어딘가에 저장해둔다) 사용자는 결과값을 가지고 무엇을 할 것인지만 정의하면 된다. 아래 코드는 두가지 차이점을 보여 준다.

Listing 4. 컬렉션에서 외부 반복작업

List<String> transactionIds = new ArrayList<>(); 
for(Transaction t: transactions){
    transactionIds.add(t.getId()); 
}

Listing 5. Stream의 내부 반복작업

List<Integer> transactionIds = 
    transactions.stream()
        .map(Transaction::getId)
        .collect(toList());

Listing 4는 명시적으로 반복문을 수행하고 거래 ID를 추출하여 리스트에 추가 하고 있다. 이와 다르게 Stream은 명시적인 반복문이 없다. Listing 5처럼 질의문의 만드는데 map은 거래 ID를 추출하고 collect는 Stream을 list로 변환해서 리턴해준다.

이제 Stream이 무엇이고 이를 가지고 할 수 있는게 무엇인지 감이 슬슬 올 것이다. Stream이 제공하는 다른 기능들에 대해서 살펴보고 실제 서비스에서 사용될만한 데이터 처리 쿼리를 작성해보자.

Stream Operations : Exploiting Streams to Process Data

Stream 인터페이스는 java.util.stream.Stream에 정의되어 있고 두가지 종류로 분류할 수 있는 많은 기능을 제공한다. Figure 1에서 봤던것 처럼 아래와 같은 기능들을 볼 수 있다.

• filter, sorted, map은 파이프라인처럼 서로 연결시킬 수 있다.
• collect는 파이프라인을 종료 시키고 결과를 리턴한다.

Stream 처리는 중간 작업(intermediate operations)라고 불리는 것을 통해 서로 연결 될 수 있다. 이들의 리턴타입은 stream이기 때문에 서로 연결될 수 있다. 이 작업은 종료 작업(terminal operations)를 통해 작업을 종료할 수 있다. 처리된 결과를 List, Integer나 void 형태로 받을 수 있다.

이를 구분해서 처리하는 것이 왜 중요한지 의문이 생길수도 있다. 중간 작업들을 Stream 파이프라인에서 종료 작업이 호출되기 전에는 절대 수행되지 않는다. 이것을 Lazy라고 한다. 

중간 작업들은 보통 병합(merged)를 통해 데이터를 처리하고 종료 작업에 이를 전달해 준다.

Listing 6.

List<Integer> numbers = Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8);
List<Integer> twoEvenSquares = 
    numbers.stream()
           .filter(n -> {
                    System.out.println("filtering " + n); 
                    return n % 2 == 0;
                  })
           .map(n -> {
                    System.out.println("mapping " + n);
                    return n * n;
                  })
           .limit(2)
           .collect(toList());

Listing 6의 코드는 주어진 숫자 목록에서 나머지가 0인 값들을 계산하는 코드이다.

아래 출력된 내용을 보면 예상과 다른 것을 볼 수 있다.

filtering 1
filtering 2
mapping 2
filtering 3
filtering 4
mapping 4

이와같은 내용으로 출력되는 원인은 limit(2)를 범위를 제한 했기 때문이다. 가끔 Stream에서 전체가 아닌 특정 부분에 대한 처리를 해야될 필요가 있다. 

지금까지 배운 Stream을 사용하는 일반적인 내용을 요약하면 3가지로 요약할 수 있다.

• 질의를 할 데이터소스(Collection)같은게 필요하다
• Stream 파이프라인을 형성하는 중간 작업
• Stream 파이프라인을 실행하고 결과를 리턴하는 종료 작업

이제 Stream에서 사용할 수 있는 기능들을 살펴보자. 이에 대한 전체 목록은 java.util.stream 패키지를 참조 하라.

Filtering : Stream에는 필터 처리를 제공하는 여러가지 메서드들이 있다.

• filter(Predicated) : 주어진 predicate(java.util.function.Predicate)와 일치하는 stream을 리턴한다.
• distinct : 중복을 제거한 유니크 엘리먼트를 리턴한다.(stream에 포함된 엘리먼트들의 equals()구현에 따라 구분된다.
• limit(n) : 주어진 사이즈(n)에 까지의 stream을 리턴한다.
• skip(n) : 주어진 엘리먼트 길이 까지 제외한 stream을 리턴한다.

Finding and matching : 가장 일반적인 데이터 처리 패턴은 주어진 조건에 일치하는 것을 찾아 내는 것이다. anyMatch,allMatch, noneMatch와 같은 기능을 사용하여 필요한 것을 찾을 수 있다. 이들을 전부 predicate를 인수로 받아서 결과를 boolean형태로 리턴한다. 에를 들어 Stream에 포함된 엘리먼트들이 모두 100보다 큰지 확인하고 싶으면 아래 코드처럼 allMatch를 사용하면 된다.

Listing 7.

boolean expensive =
    transactions.stream()
        .allMatch(t -> t.getValue() > 100);

Stream은 임의의 요소를 찾는 findFirst, findAny 기능도 제공한다. 이 기능은 필터와 같이 Stream을 조작하는데 사용될 수 있다.FindFirst, FindAny 값은 Optional 객체를 리턴한다.

Listing 8.

Optional<Transaction> = 
    transactions.stream()
        .filter(t -> t.getType() == Transaction.GROCERY)
        .findAny();

Optional<T> 클래스(java.util.Optional)는 값이 존재 하거나 존재하지 않을 경우에 사용할 수 있는 컨데이터다. Listing 8의 크도를 보면 Transaction 타입이 GROCERY가 아닌것이 있을 수도 있다. Optional 클래스는 엘리먼트들이 존재하는지 확인하는 여러 메서드를 가지고 있다. 예를 들어 만약 엘리먼트가 존재 한다면 ifPresent 메서드를 사용해서 특정 작업을 수행할 수 있다.

Listing 9.

transactions.stream()
    .filter(t -> t.getType() == Transaction.GROCERY)
    .findAny()
    .ifPresent(System.out::println);

Mapping : map 기능을 사용해서 인자(java.util.function.Function)로 전달받은 값들을 가지고 새로운 stream을 만들 수 있다. 

예를 들어 Stream에 포함된 엘리먼트들에 대한 특정 정보만 필요할 수도 있다. Listin 10 처럼 List에 저장된 단어의 길이들을 가지고 있는 새로운 List를 생성할 수 있다. 

Listing 10.

List<String> words = Arrays.asList("Oracle", "Java", "Magazine");
List<Integer> wordLengths = words.stream()
    .map(String::length)
    .collect(toList());

Recuding : 지금까지 본 stream의 종료작업은 boolean(allMatch 등), void(forEach), optional(findAny 등) 오브젝트들을 결과로 리턴했고 collect를 사용해서 List형태로도 리턴을 할 수 있었다.

이외에도 "가장 큰 ID를 가진 거래내역은 무엇인가", "모든 거래 내역의 합계" 처럼 Stream의 엘리먼트들에 복잡한 형태의 공식같은걸 적용하는 처리 쿼르를 만들수도 있다. 이 기능은 Stream의 reduce를 통해서 사용한데 이는 결과가 생성될때 까지 반복적으로 각 요소에 대한 (두 숫자를 추가 한다든지)동작을 허용한다.

이것을 함수형 프로그래밍에서는 고차 함수(fold operation)라고 부른다.

코드들을 보면 좀 더 쉽게 이해할 수 있을 것이다.

for 반복문을 사용해서 덧셈을 할때는 아래와 같은 코드로 작성된다.

int sum = 0;
for (int x : numbers) {
    sum += x; 
}

각각 엘리먼트드을 순회하면서 값들을 더하도록 되어 있다. 이 코드에는 두개의 파라미터가 있다고 볼 수 있는데 sum( 여기서는 0)은 값을 표현하는 것이고, 각 엘리먼트들을 더하귀 위한 것(여기서는 +)가 있다.

Stream에서 reduce 메서드를 사용한다면 엘리먼트들의 합은 Listing 11과 같은 코드로 표현된다.

reduce메서드는 두개의 인자값을 가진다. 

Listing 11.

int sum = numbers.stream().reduce(0, (a, b) -> a + b);

• 초기 화 값, 0
• BinaryOperator<T>, 두개의 엘리먼트들을 더해서 새로운 값을 생성

reduce는 반복 적용 패턴을 추상화 해준다. 모든 값을 곱한다던지 최대값을 구하는 것들은 아래처럼 표현될 수 있다.

Listing12.

int product = numbers.stream().reduce(1, (a, b) -> a * b);
int product = numbers.stream().reduce(1, Integer::max);

Numeric Streams 

reduce 메서드를 사용해서 Stream에 있는 숫자형 값들에 대한 합계를 구할 수 있다. 하지만 이를 계산하는데 오브젝트를 Integer 형태로 Boxing 처리를 해야 하기 때문에 어느 정도 Cost가 든다. 아래 코드 처럼 명시적으로 sum() 메서드를 호출하는 것이 좀 더 좋아 보인다.

Listing 13.

int statement = 
    transactions.stream()
        .map(Transaction::getValue)
        .sum(); // error since Stream has no sum method

위와 같은 기능을 제공하기 위해 Stream에서는 3가지 기본형태의 Stream 인터페이스를 제공한다. IntStream, DoubleStream,LongStream 이 Stream들은 int, double, long 형태의 Stream을 처리 하도록 되어 있다.

가장 빈번히 사용될 메서드는 아마 mapToInt, mapToDouble, mapToLong일 것이다 이 3가지 메서드들을 이전에 보았던 map과 동일한 동작을 하지만 이것의 리턴값은 Stream<T> 형태의 값을 준다. 예를 들어 Listing 13의 코드는 Listing 14처럼 수정할 수 있다. 

Listing 14.

int statementSum = 
    transactions.stream()
        .mapToInt(Transaction::getValue)
        .sum(); // works!

마지막으로 숫자형 Stream의 다른 활용 방법은 숫자 범위를 지정할 수 있는데 있다. 예를 들어 1에서 100사이의 값을 생성하려고 할때 range와 rangeClosed 메서드를 IntStream, DoubleSteam, LongStream에서 사용할 수 있다.

이 두 메서드는 첫번째 인자와 마지막 인자사이의 값을 가져온다. range는 마지막 값을 제외하고 계산하고 rangeClosed는 이를 포함해서 계산한다. Listing 15에서 rangeClosed를 사용해서 10과 30사의 홀수를 가져오는 것을 볼 수 있다.

Listing 15.

IntStream oddNumbers = 
    IntStream.rangeClosed(10, 30)
        .filter(n -> n % 2 == 1);

Building Streams

Stream을 생성하는 몇가지 방법들이 있는데 지금까지는 컬렉션을 통해서 Stream을 생성했었고, 숫자를 사용해서 만들기도 했다. Stream은 값 목록이나 배열, 파일과 같은곳에서도 생성할 수 있으며 무한 Stream을 생성하기 위해 함수를 사용할 수도 있다.

값 목록이나 배열에서 Stream을 생성하는 것은 어렵지 않다. 스태틱 메서드인 Stream.of를 사용하거나 Arrays.Stream을 사용하면 된다.

Listing 16.

Stream<Integer> numbersFromValues = Stream.of(1, 2, 3, 4);
int[] numbers = {1, 2, 3, 4};
IntStream numbersFromArray = Arrays.stream(numbers);

파일에서 Stream을 생성할 경우 Files.Lines를 사용하면 되고, 아래 코드는 파일에서 라인수를 카운팅한다.

Listing 17.

long numberOfLines = 
    Files.lines(Paths.get(“yourFile.txt”), Charset.defaultCharset())
    .count();

Infinite streams : 지금까지 Stream이 무엇이고 어떻게 사용하는지에 대해서 이래했을 것이다. 함수형 프로그래밍에서 Stream을 생성하는 방법은 Stream.iterate와 Stream.generate가 있다. 엘리먼트들이 필요한 시점에 계산되기 때문에 이 두 함수는 지속적으로 엘리먼트들을 가져 올 수 있다. 이것이 바로 무한 Stream이다(고정된 사이즈의 Stream이 아닌)

Listing 18은 iterate를 사용해서 모든 숫자에 10을 더하는 기능이다. iterate메서드는 초기값 0부터 시작해서 계속 새로운 값을 생성한다. 

Listing 18.

Stream<Integer> numbers = Stream.iterate(0, n -> n + 10);

limit()을 사용해서 무한 Stream에서 고정된 크기의 Stream을 생성할 수 있다. Listing 19는 Stream의 사이즈를 5로 고정한 것을 보여주는 코드이다.

Listing 19.

numbers.limit(5).forEach(System.out::println); // 0, 10, 20, 30, 40

결론.

Java SE 8에서는 빠르고 정교한 데이터 처리를 위해 Stream API를 공개했다. 이 문서에서는 Stream이 제공하는 많은 기능들(filter,map, reduce, iterate)을 살펴 봤고 이를 활용해서 간결한 코드로 정교한 데이터 처리 작업을 할 수 있었다. 이 방법은 Java SE 8 이전에 Collection 데이터들을 처리하는 방법과 매우 다르지만 많은 장점이 있다.

1. Steream API는 몇가지 기능들을 활요해서 laziness 및 short-circuiting을 통해 데이터 처리를 최적화 할 수 있다.
2. Stream은 병렬 처리를 손쉽게 지원한다.

---

이전글에 이어 2부를 번역해 봤습니다.

원본 : Part 2. Processing Data with Java SE 8 Streams

==================================================================================

다양한 기능들을 조합하여 좀 더 다양한 데이터 처리 쿼리를 만들기.

이전 문서에서 스트림을 사용하여 컬렉션 데이터 처리를 데이터베이스의 쿼리처럼 처리하는것을 살펴 보았다.
머 리를 식힐겸 스트림 API를 사용하여 거래금액이 비싼 내역들의 합을 구하는 코드인 Listing 1을 보자. 중간 작업(filter, map)을 통해 파이프라인을 세팅하고 종료 작업(reduce)을 통해 코드를 실행시키는 과정을 아래 Figure 1을 통해 볼 수 있다.

Listing 1.

int sumExpensive = transactions.stream()
                   .filter(t -> t.getValue() > 1000)
                   .map(Transaction::getValue)
                   .reduce(0, Integer::sum);

Figure 1

flatMap, collect 두 기능에 대해서 알아보자.

• flatMap : 중간 작업으로 "map" 과 "flatten"을 섞은 것이다
• collect : 종료 작업으로 인자로 받은 형태의 데이터 타입으로 결과물을 생성한다. 
  

이 두 기능은 복잡한 형태의 질의를 하는데 유용하게 쓰인다. 예를 들어 flatMap과 collect를 조합하여 Listing2 처럼 스트림에서 중복이 제거된 각 알파벳의 카운트를 가진 맵을 만들 수 있다. 
이 두 기능에 대해서 더 자세히 살펴보고 설명할 예정이니 처음 보는 형태의 코드를 보고 걱정할 필요는 없다

Listing 2

import static java.util.function.Function.identity;
import static java.util.stream.Collectors.*;

Stream<String> words = Stream.of("Java", "Magazine", "is", "the", "best");

Map<String, Long> letterToCount =
           words.map(w -> w.split(""))
                .flatMap(Arrays::stream)
                .collect(groupingBy(identity(), counting()));

Listing2 의 결과를 출력해보면 Listing3과 같이 표시된다. 

이제부터 flatMap과 collect가 어떻게 동작하는지 알아보자

Listing 3

[a:4, b:1, e:3, g:1, h:1, i:2, ..]

flatMap Operation
만약 파일에서 고유한 단어를 찾는다고 생각해보자. 이를 어떻게 구현할 것인가?

Files.lines()를 통해 한줄씩 읽은 다음 map() 기능을 사용해서 단어들을 잘라낸 다음에  distinct()를 사용해 중복을 제거 하면 된다. 이를 코드로 작성해보면 Listing 4처럼 될 것이다.

Listing 4

Files.lines(Paths.get("stuff.txt"))
              .map(line -> line.split("\\s+")) // Stream<String[]>
              .distinct() // Stream<String[]>
              .forEach(System.out::println);

하지만 이 코드는 원하는 대로 동작 하지 않는다. 이 코드를 실행시켜보면 아래와 같은 결과를 얻을 것이다.

[Ljava.lang.String;@7cca494b
[Ljava.lang.String;@7ba4f24f

처음 작성한 위 코드는 스트림들을 String으로 출력하고 있다. 무슨일이 발생한걸까? 

이 방식의 문제점은 람다 표현식에서 map으로 전달되는 형태가 String 배열(String[]) 이기 때문이다.
이 결과를 보면 우리가 원하는 형태는 문자들로 구성된 Stream<String>인데 map()을 사용했을때 리턴하는 스트림은 Stream<String[]>이 된다.

이는 flatMap을 사용하면 간단히 해결할 수 있다. 이 방법을 한단계씩 적용해보자.

코드를 작성하기 위해서 단어의 스트림 대신 배열 스트림이 필요하다. Arrays.stream() 메서드를 통해 배열을 스트림으로 만들 수 있다.  

Listing 5

String[] arrayOfWords = {"Java", "Magazine"};
Stream<String> streamOfwords = Arrays.stream(arrayOfWords);

이처럼 배열을 스트림으로 변환하는 내용을 Listing 6 처럼 추가해보자.
이 방법도  스트림의 스트림목록 형태(Stream<Stream<String>>)로 되어 있어서 잘 동작하지 않는다. 
처음엔 매 라인을 단어들의 배열로 바꾸도록 하였고, 그 다음 이 배열들을 Arrays.stream()을 사용해서 stream으로 변경했다.

Listing 6

Files.lines(Paths.get("stuff.txt"))
            .map(line -> line.split("\\s+")) // Stream<String[]>
            .map(Arrays::stream) // Stream<Stream<String>>
            .distinct() // Stream<Stream<String>>
            .forEach(System.out::println);

Listing 7 처럼 flatMap을 사용하도록 하여 잘 동작하게 할 수 있다.  이는 map(Arrays::stream)을 사용한것 처럼 각각 분리된 스트림들을 생성하고 이를 하나의 스트림으로 변경한다. 

Listing 7

Files.lines(Paths.get("stuff.txt"))
            .map(line -> line.split("\\s+")) // Stream<String[]>
            .flatMap(Arrays::stream) // Stream<String>
            .distinct() // Stream<String>
            .forEach(System.out::println);

Figure 2 에서는 flatMap 메서드의 동작방식을 설명하고 있다.

Figure 2

간단하게 말하면 flatMap은 각각 다른 값을 가진 스트림들을 하나의 스트림으로 생성해 준다. 
flatMap은 자주 사용되는 패턴으로 Optional이나 CompletableFuture를 사용할때 다시 보게 될 것이다.

collect Operation
이젠 colllect 메서드에 대해서 더 상세히 알아보자. 이전 문서에서 봤던 collect의 사용방법은 작업을 처리한 후 다른 스트림을 리턴하거나 다른 값(boolean, int, Optional)을 리턴하는 것이었다.

collect 메서드는 종료 작업이긴 하지만 stream을 list로 변경할 경우는 약간 다르다. 예를 들어 아래 listing 8의 코드 처럼 거래 금액이 큰 거래내역의 id 목록을 가져오는것 처럼 list를 리턴한다.

Listing 8

import static java.util.stream.Collectors.*; 
List<Integer> expensiveTransactionsIds = 
       transactions.stream()
                   .filter(t -> t.getValue() > 1000)
                   .map(Transaction::getId)
                   .collect(toList());

collect 메서드에 전달된 인자의 타입은 java.util.stream.Collector다. Collector 오브젝트가 하는일은 무엇일까? 이것은 본질적으로 최종 리턴값으로 되어야할 것에 대해 설명을 한다고 보면 된다.
팩토리 메서드인 Collectors.toList()는 스트림을 리스트 형태로 변경하여 반환한다. 다른 형태의 내장 Collectors도 여러가지가 존재 한다.

스트림을 다른 컬렉션 형태로 변경하기.
toSet()을 사용하면 스트림을 Set(중복이 제거된)으로 변경 가능하다. Listing 9의 코드는 거래내역중에 도시 목록을 중복이 제거된 Set 형태로 반환한다. 
Listing 9

Set<String> cities = 
         transactions.stream()
                     .filter(t -> t.getValue() > 1000)
                     .map(Transaction::getCity)
                     .collect(toSet());

위 코드를 보면 Set이 어떤 타입인지는 보장하지 않지만 toCollection() 메서드를 사용해서 타입을 지정할 수 있다.
아래 Listing 10의 코드 처럼 toCollecte에다가 HashSet을 지정할 수 있다.

Listing 10

Set<String> cities = 
         transactions.stream()
                     .filter(t -> t.getValue() > 1000)
                     .map(Transaction::getCity)
                     .collect(toCollection(HashSet::new));

위 기능은 collect를 사용해서 할 수 있는 것들중에 작은 부분일 뿐이다. 
아래 예제처럼 몇가지 기능을 더 사용할 수 있다.

• 화폐에 따른 거래 내역을 그룹핑하여 합을 구하기 (Map<Currency, Integer>)
• 거래 내역을 두 개로 분리 : 비싼 내역과 그렇지 않은 내역(Map<Boolean, List<Transaction>)
• 다양한 뎁스로 그룹핑, 도시 형태로 먼저 그룹핑 하고 그 안에 비싼 내역과 그렇지 않은 내역으로 또 그룹핑(Map<String, Map<Boolean, List<Transaction>>)
  

계속 해서 Steam API와 collectors 들을 사용해서 다양한 질의를 생성하는것에 대해 살펴보자
가 장 먼저 볼 것은  스트림을 "summarizes" 하는 것이고(평균값 계산, 최대 및 최소값 추출) 그 다음은 간단한 그룹핑을 살펴보고 마지막으로 collectors 들을 조합하여 멀티레벨 그룹핑 같은 다양한 질의문을 만들어보자.

Summarizing
몇가지 간단한 예제들을 가지고 워밍업을 해보자. 
이전 문서에서 숫자들의 목록을 가지고 최대, 최소 및 평균을 reduce 메소드를 사용해서 구해봤었다. 이것들을 미리 정의 되어 있는 collectors 들이다. 
만약 숫자들의 갯수들 가져오고 싶으면 Listing 11의 코드처럼 counting()을 사용하면 된다.

Listing 11

long howManyTransactions =  transactions.stream().collect(counting());

summingDouble(), summingInt(), summingLong() 같은  메서드를 사용해서 스트림에 포함된 Double, Int, long 과 같은 엘리먼트들의 합계를 구할 수 있다. 
아래 Listing 12의 코드는 모든 거래 내역의 합계를 구한다.

Listing 12

int totalValue = transactions.stream()
           .collect(summingInt(Transaction::getValue));

위와 비슷하게 averagingDouble(), averagingInt(), averagingLong()일 사용해서 평균값을 구할 수 있다.

maxBy(), minBy()를 사용해서 최대값과 최소값을 구할 수도 있다.
위 두 메서드는 Figure 3처럼  Comparator형태의 인자를 필요로 한다.
Figure 3

Listing 14의 코드 처럼 스태틱 메서드인 comparing() 메서드는 전달된 함수를 가지고 Comparator 객체를 생성한다. 이 함수에서 추출된 값을 사용해서 스트림의 값들을 비교하게 된다. 
아래 코드에서는 가장 높은 값을 찾는 코드이다.

Listing 14

Optional<Transaction> highestTransaction = 
        transactions.stream()
                               .collect(maxBy(comparing(Transaction::getValue)));

reducing() 메서드 처럼 모든 엘리먼트들에 대해 정해진 작업을 반복적으로 수행하는 것도 있다. 
Listing 15의 코드는 reducing() 메서드를 사용해서 합계를 구하는 코드이다.
Listing 15

int totalValue = transactions.stream().collect(reducing(
        0, Transaction::getValue, Integer::sum));

recuding() 메서드는 3개의 인자를 가진다

• 초기화 값(stream이 비어있으면 이겂을 리턴) : 여기서는 0
• 스트림에서 반복적으로 수행할 함수 : 여기서는 getValue를 통해 값을 추출
• 위에서 구한 값을 조합 하는 코드 : 여기서는 값을 계속 더한다.
  

"이전에 reduce(), max(), min()을 통해서 위 기능을 사용했었는데 왜 이제와서 이걸 보여주지?' 라는 의문이 들 것이다. 
위의 기본 기능이외에 collectors를 조합하여 더 복잡한 질의문(합계를 그룹핑 하는)을 생성하는것을 볼것인데 이 내장 collector에 대해서 알아 두는 것이 좋다.

Grouping
대부분의 데이터베이스들이 Group By 같은 명령어를 통해서 데이터 그룹핑을 지원한다.

만약 화폐에 따라 거내 내역을 그룹핑할 필요가 있을때 이를 이전 방법으로 작성한다면 아래 코드처럼 고통스러운 코드를 작성해야 될 것이다.

Listing 16

Map<Currency, List<Transaction>> transactionsByCurrencies = new HashMap<>();
for(Transaction transaction : transactions) { 
        Currency currency = transaction.getCurrency();
        List<Transaction> transactionsForCurrency = transactionsByCurrencies.get(currency);
        if (transactionsForCurrency == null) {
                transactionsForCurrency = new ArrayList<>();
                transactionsByCurrencies.put(currency, transactionsForCurrency);
        }
        transactionsForCurrency.add(transaction);
}

위 코드를 보면 먼저 거래내역을 저장할 Map을 생성한 후 거래내역에 대해 반복 작업을 수행해서 currency를 추출 한 다음 Map에 currency가 존재 하지 않으면 새로 만들어서 넣어 주는 작업을 하고 있다.
사실 currency에 따라 거래내역을 그룹핑 하는것을 원할 뿐인데 상당히 복잡한 코드로 작성되어 있다. 
이를 개선하기 위해서는 groupingBy()메서드를 사용하면 되는데 이를 사용해서 간결한 코드로 위 내용을 작성할 수 있다.
Listing 17의 코드는 이전 코드와 동일한 역할을 하고 훨씬더 이해하기 쉽게 작성되어 있다.
Listing 17

Map<Currency, List<Transaction>> transactionsByCurrencies =
    transactions.stream().collect(groupingBy(Transaction::getCurrency));

팩토리 메서드인 groupingBy()는 그룹핑 하는데 키로 사용할 값을 추출하는 함수를 인자로 받아서 사용한다. 이를 classification function(분류함수)이라고 한다. 
이 예제에서는 transaction의 currency로 그룹핑을 하기 위해서 메소드 참조(Transaction::getCurrency)를 파라미터로 전달한다. 
Figure 4에서는 이를 그림으로 설명하고 있다.
Figure 4

Partitioning
groupingBy()의 특별한 케이스인 partitioningBy()라는 팩토리 메서드도 존재 한다. 이것은 predicate타입을 인자로 받으며 predicate에 만족하는 엘리먼트들을 그룹핑해서 리턴한다. 
바꿔 말하면 거래내역들을 분류해서 그룹핑 할 수 있다(Map<Boolean, List<Transaction>)
예 를 들어 만약 거래내역을 두가지 그룹으로 나누고 싶다면(고가, 저가) Listing 18의 코드처럼 partitioningBy()를 사용하면 된다. 람다 표현식인 t -> t.getValue() > 1000 을 기준으로 거래 내역을 구분하게 된다.
Listing 18

Map<Boolean, List<Transaction>> partitionedTransactions =
            transactions.stream().collect(partitioningBy(
              t -> t.getValue() > 1000));

Composing collectors.
SQL을 자주 사용해봐서 익숙 하다면 GROUP BY와 함께 COUNT()와 SUM()을 사용해본 경험이 있을 것이다. 이 기능과 비슷한것을 Stream API를 통해서 할 수 있다. 
groupingBy() 두번째 인자로 대상 을 넣어 주면 된다. 

추 상적인 설명으로는 이해가 잘 않으니까 간단한 예제를 살펴보자. Map을 만들어서 도시별로 거래내역의 합계를 구한다고 해보자. Listing 19의 코드 처럼 groupingBy 의 첫번째 인자로 getCity()를 줘서 키를 지정하면 도시를 키로 가지고 있고 값으로 List<Transaction>을 가지고 있는 Map을 리턴할 것이다.

Listing 19

Map<String, Integer> cityToSum = 
           transactions.stream().collect(groupingBy(
             Transaction::getCity, summingInt(Transaction::getValue)));

만약 도시별 거래금액의 합계를 구하기 위해서는 summingInt라는 추가적인 컬렉터를 세팅해줘야된다. 이 코드의 실행 결과로 Map<String, Integer>를 형태로 각 도시별 거래금액의 합계를 구할 수 있다.
groupingBy(Transaction::getCity> 형태로 사용할 수도 있고 groupingBy<Stransaction::getCity, toList()) 형태로도 사용할 수 있다.

각 도시의 가장 높은 거래 금액을 가지고 있는 Map을 생성한다고 해보자. 
아래 Lisinting 20의 코드처럼 이전에 봣던 maxBy 컬렉터를 사용하면 쉽게 구현할 수 있다.
Listing 20

Map<String, Optional<Transaction>> cityToHighestTransaction = 
           transactions.stream().collect(groupingBy(
             Transaction::getCity, maxBy(comparing(Transaction::getValue))));

스트림 API는 다양한 기능을 제공하고 SQL처럼 동작하는 쿼리를 간결하게 작성할 수 있다. 

마지막으로 조금 더 복잡한 예를 살펴보자. groupingBy()의 인자로 다른 컬렉터를 받아서 다양한 작업을 할 수 있다. 
이것은 groupingBy()도 collector이기 때문이다. 멀티레벨 그룹핑을 할때 groupingBy collector를 전달함으로써 이를 구현할 수 잇다.

Listing 21의 코드는 거래 내역을 도시별로 그룹핑하고 하위 그룹은 통화별 거래내역의 평균을 가지도록 하는 코드이다.

Listing 21

Map<String, Map<Currency, Double>> cityByCurrencyToAverage = 
           transactions.stream()
               .collect(groupingBy(Transaction::getCity,
                   groupingBy(Transaction::getCurrency,  
                   averagingInt(Transaction::getValue))));

Figure 5는 이름 그림으로 설명한 것이다.

Creating your own collector.
지금까지 본 모든 collector 들은 java.util.stream.Collector 인터페이스를 구현한 것이다. 만약 자신만의 collector를 구현하고 싶으면 이 인터페이스를 상속 받아서 만들 면 된다. 

결론
이 문서에서는 Stream API의 두가지 쓸만한 기능인 flatMap과 collect에 대해서 살펴봤다. 이를 사용해서 데이터 처리 작업을 간결한 코드로 할 수 있었다.
특 히 collect 메서드는 summarizing, grouping, partitioning을 쉽게 만들 수 있고 이를 서로 조합해서 "각 도시의 통화별 거래내역의 합계를 가지고 있는 두 단계 깊이의 Map 처럼 유용한 쿼리를 만들 수도 있다.

이 문서에 collector들에 대해 전부 알아보지는 않았기 때문에 다른 Collectors들(mapping(), joining(), collecting AndThen()) 같은 것들도 살펴보면 쓸만할 것이다.

출처: http://starplatina.tistory.com/