Java 대용량 데이터 처리 (2) - Stream 메모리 효율 높이기

2. List -> Stream (장기간 저장된 로그 분석)

Stream 처리 예제

Bad case : 모두 조회, 모두 처리

List<Log> logs = logRepository.findAll(); // 100만 건
logs.forEach(this::analyze);

• OutOfMemoryError 발생 위험
• GC 부하
• 응답속도 지연

Stream 처리란?

데이터를 한 번에 다 불러오지 않고, 하나씩 처리하면서 다음 데이터를 읽는 방식

• List에 모두 올리지 않기
• Iterator처럼 지연(Lazy) 처리

Good case : stream으로 하나씩 처리

@QueryHints(value = @QueryHint(name = org.hibernate.jpa.QueryHints.HINT_FETCH_SIZE, value = "1000"))
@Query("SELECT l FROM Log l")
Stream<Log> findAllByStream();

try (Stream<Log> stream = logRepository.findAllByStream()) {
    stream.forEach(this::analyze);
}

// 외부 파일 읽는 경우
try (BufferedReader reader = new BufferedReader(new FileReader("importFile.csv"))) {
    reader.lines().forEach(this::parseLine);
}

주의할 점

• try-with-resources 를 통해 자원 누수 관리 할 것
• 가비지 컬렉션 고려
• 병렬 충돌 방지할 것 (parallelStream() 등)

연재물(?)이기 때문에 cursor를 보고 stream을 보면 유사하다는 생각이 든다.

하지만
Cursor는 DB에 WHERE 조건을 계속 바꿔가며 반복 조회하는 방식이고,
Stream은 한 번 조회된 결과 집합을 순차적으로 흘려보내며 처리하는 방식이기 때문에

근본적인 관점과 적용 시점이 다르다.

Cursor : 처리 흐름을 세밀하게 제어할 수 있으며, 병렬 배치 처리에 유리하다
Stream : 성능을 극대화할 수 있으며, 코드가 간단하고 직관적이다

 

그렇다면 함께 쓸 수도 있지 않을까? 그렇다

• 전체 데이터를 반복 쿼리로 나눠서 처리하되
• 각 쿼리 결과가 많아서 메모리에 부담이되고
• 데이터의 처리도 빠르게 이루어졌으면 싶을때

* Repository

@Query(value = "SELECT * FROM data_table WHERE id > :lastId ORDER BY id ASC LIMIT :limit", nativeQuery = true)
Stream<Data> streamBatch(@Param("lastId") Long lastId, @Param("limit") int limit);

Stream<T>을 리턴하면 JPA는 내부적으로 ResultSet을 열어둔 상태, try-with-resources 를 반드시 활용할 것

public void processInBatches() {
    Long lastId = 0L;
    int batchSize = 1000;

    while (true) {
    	// try-with-resources
        try (Stream<Data> batchStream = repository.streamBatch(lastId, batchSize)) {
            List<Data> currentBatch = new ArrayList<>();

            batchStream.forEach(data -> {
                process(data);                 
                currentBatch.add(data);
            });

            if (currentBatch.isEmpty()) break;

            lastId = currentBatch.get(currentBatch.size() - 1).getId(); // 커서 갱신
        }
    }
}

• 메모리 효율 증가 : Stream이므로 한 줄씩 처리하고 GC의 대상이 됨
• 흐름 제어 : Cursor로 반복 쿼리 가능
• 정렬 보장 : Cursor의 id asc 기준으로 순차처리
• 예측 가능성 : 배치 단위 처리로 로그, 모니터링에도 유리함(어디까지 처리했는지, 어떤 데이터에서 실패했는지 등)

 

요약 정리

1. Cursor + Stream 방식은 반복 쿼리로 흐름을 제어하면서, 각 배치 내에서는 Stream으로 메모리 효율을 극대화할 수 있음
2. 일반 Cursor 방식보다 메모리 점유가 낮고, 순차적 처리에서는 가장 안정적인 대용량 처리 구조
3. Stream은 try-with-resources로 닫아줘야 하며, 커서 추적과 예외 처리를 명확히 설계해야 안정적