TIL

JPA ItemReader / ItemWriter

JpaCursorItemReader

• JPA에서 사용하는 커서 방식 ItemReader
• JpaCursorItemReader는 내부적으로 entityManager를 통해 데이터를 읽는다.

JpaCursorItemReader 해부

JpaCursorItemReader
    │
    ├────── queryString(JPQL) or JpaQueryProvider  
    │        └─ (Query 생성에 사용됨)  
    │
    ├────── EntityManager  
    │        └─ (JPA 핵심 엔진)
    │        
    └────── Query  
             └─ (EntityManager가 생성하는 실행 가능한 쿼리 인스턴스)

• queryString
  • JpaCursorItemReaderBuilder.queryString() 메서드를 통해 생성
  • JpaCursorItemReaderBuilder.queryProvider()로 커스텀 JpaQueryProvider 전달 가능
• EntityManager
  • JpaCursorItemReaderBuilder.entityManagerFactory()로 EntityManagerFactory를 전달하면 내부적으로 EntityManager를 생성해 사용된다.
• Query
  • JpaCursorItemReader가 EntityManager로 Query 객체를 생성
  • 스트리밍 방식으로 getResultStream() 메서드를 호출
• 실행 흐름
  • JpaCursortemReader 초기화
    • doOpen() 호출
    • Query 객체 생성
    • getResultStream() 호출하여 커서를 순회할 Iterator 준비
  • Iterator를 통해 실제 데이터를 한 건씩 읽어온다.
    • iterator.hasNext()로 데이터가 있다면 iterator.next()로 데이터 반환
    • 더 이상 읽을 데이터가 없다면 null 반환, 읽기 종료

JpaCursorItemReader 구성 예제

@Bean
@StepScope
fun postBlockReader(
    @Value("#{jobParameters['startDateTime']}") startDateTime: LocalDateTime,
    @Value("#{jobParameters['endDateTime']}") endDateTime: LocalDateTime,
): JpaCursorItemReader<Post> =
    JpaCursorItemReaderBuilder<Post>()
        .name("postBlockReader")
        .entityManagerFactory(entityManagerFactory)
        .queryString(
            """
                SELECT p FROM Post p JOIN FETCH p.reports r
                WHERE r.reportedAt >= :startDateTime AND r.reportedAt < :endDateTime
            """.trimIndent()
        )
        .parameterValues(
            mapOf(
                "startDateTime" to startDateTime,
                "endDateTime" to endDateTime
            )
        )
        .build()

• entityManagerFactory()
  • 빈 주입 받은 EntityManagerFactory 구성
• queryString()
  • parameterValues로 받은 파라미터와 함께 쿼리 정의
  • 이 메서드를 사용하면 내부적으로 JpaQueryProvider 구현체가 사용된다.

JpaQueryProvider를 사용한 쿼리 설정

• JpaQueryProvider 구현체로 더 유연한 쿼리 처리가 가능하다.
  • JpaNamedQueryProvider: 엔티티 등에 미리 정의된 Named Query를 사용
  • JpaNativeQueryProvider: Native SQL을 사용하여 데이터를 조회
  • 필요하다면 커스텀 구현체도 등록할 수 있다.

// JpaNamedQueryProvider 예제

@Entity
@Table(name = "posts")
@NamedQuery(
    name = "Post.findByReportsReportedAtBetween",
    query = "SELECT p FROM Post p JOIN FETCH p.reports r WHERE r.reportedAt >= :startDateTime AND r.reportedAt < :endDateTime",
)
class Post() { ... }

@Bean
@StepScope
fun postBlockReader(
    @Value("#{jobParameters['startDateTime']}") startDateTime: LocalDateTime,
    @Value("#{jobParameters['endDateTime']}") endDateTime: LocalDateTime,
): JpaCursorItemReader<Post> =
    JpaCursorItemReaderBuilder<Post>()
        .name("postBlockReader")
        .entityManagerFactory(entityManagerFactory)
        .queryProvider(createQueryProvider())
        .parameterValues(
            mapOf(
                "startDateTime" to startDateTime,
                "endDateTime" to endDateTime
            )
        )
        .build()
        
private fun createQueryProvider(): JpaNamedQueryProvider<Post> {
    val queryProvider = JpaNamedQueryProvider<Post>()
    queryProvider.setEntityClass(Post::class.java)
    queryProvider.setNamedQuery("Post.findByReportsReportedAtBetween")
    return queryProvider
}

JpaPagingItemReader

• JPA 구현체를 사용한 페이지 단위 쿼리 ItemReader

offset 기반 페이징

• JpaPagingItemReader는 JdbcPagingItemReader와 달리 offset 기반으로 페이징을 수행한다.
  • ex) SELECT * FROM victims ORDER BY id LIMIT 10 OFFSET 0
• offset 기반 페이징은 다음 문제들이 존재한다.
  • 데이터 정합성 붕괴
    • offset 페이징 도중 데이터가 변경되면 쿼리 결과에 중복이 포함되거나 누락이 발생할 수 있다.
  • 매우 느린 성능
    • offset 기반으로 데이터를 읽으면 필요한 위치까지 모두 읽을 필요가 있다.
    • ex) offset 999,900이라면 이 쿼리에선 999,900개를 읽고 버리고 마지막 limit만큼을 반환한다.
  • 메모리 고갈
    • DB가 전체 결과를 메모리에 로딩하고 필요한 부분을 자르는 방식을 사용한다.
• 다만 작은 데이터셋을 대상으로 하는 상황에선 실용적인 선택이 될 수도 있다.

JpaPagingItemReader 해부

JpaPagingItemReader 
    │ 
    ├────── queryString(JPQL) or JpaQueryProvider 
    │        └─ (Query 생성에 사용됨) 
    │ 
    ├────── EntityManager 
             └─ (JPA 핵심 엔진)

• JpaCursorItemReader와 비슷하지만 읽기 방식에 근본적인 차이가 있다.
  • 초기화 시점에 한 번의 쿼리를 실행하는 JpaCursorItemReader와 달리 매 doReadPage() 메서드마다 새로운 쿼리를 생성하고 실행한다.
  • 각 쿼리는 offset에 따라 getResultList()를 통해 페이지 데이터를 가져온다.

JpaPagingItemReader 예제

@Bean
@StepScope
fun postBlockReader(
    @Value("#{jobParameters['startDateTime']}") startDateTime: LocalDateTime,
    @Value("#{jobParameters['endDateTime']}") endDateTime: LocalDateTime,
): JpaPagingItemReader<Post> =
    JpaPagingItemReaderBuilder<Post>()
        .name("postBlockReader")
        .entityManagerFactory(entityManagerFactory)
        .queryString(
            """
            SELECT DISTINCT p FROM Post p
            JOIN p.reports r
            WHERE r.reportedAt >= :startDateTime AND r.reportedAt < :endDateTime
            ORDER BY p.id ASC
            """.trimIndent(),
        ).parameterValues(
            mapOf(
                "startDateTime" to startDateTime,
                "endDateTime" to endDateTime,
            ),
        ).pageSize(5)
        .build()

• pageSize()
  • 한 번 쿼리로 읽어올 페이지 크기를 결정 (LIMIT 절의 값으로 사용)
• queryString() 수정
  • JpaCursorItemReader 때와 다르게 fetch join 제거했다.
    • Post -* Report 관계처럼 일대다 관계에서 페치 조인과 limit/offset 페이징을 함께 사용하면 메모리 낭비가 발생한다.
    • 쿼리에서 페이징을 하는게 아닌, 전 데이터를 메모리에 올리고 페이징을 하기 때문
    • 그리고 페치 조인 제거로 인한 n + 1를 또 방지하기 위해선 @BatchSize를 적용하면 지정된 수만큼 IN 절로 조회하는 최적화를 사용할 수 있다. (@BatchSize(size = 5))
  • ORDER BY 추가
    • ORDER BY가 없으면 읽을 때마다 데이터 순서가 보장되지 않아 누락과 중복이 발생할 수 있다.

일대다 관계에서의 BatchSize 사용 시, FetchType을 EAGER로 변경해야 IN 절 쿼리를 통한 일괄 조회가 가능하다. 이는 JpaPagingItemReader에서 데이터를 읽고, 실제로 Report 객체가 필요한 ItemProcessor에 넘기기 전에 커밋을 해버리기 때문이다. BatchSize는 트랜잭션 범위 내에서만 동작하므로 이 시점에선 N+1이 발생한다. 때문에 EAGER 전략을 통해 ItemReader 트랜잭션 안에 있을 때 미리 페치를 끝내놔야 한다.

transacted 필드와 시스템 안정성 확보

• 앞서 설명한 BatchSize가 무효화되는 문제는 JpaPagingItemReader에서 transacted = true이기 때문에 발생한다.
• transacted가 true일 때 JpaPagingItemReader는 페이지를 읽기 전에 entityManager.flush()를 호출한다.
  • ItemProcessor에서 엔티티 변경이 생긴다면 다음 데이터를 읽어들이는 doReadPage() 시점에 의도치 않은 DB 변경이 발생할 수 있다.

if (transacted) {
    tx = entityManager.getTransaction();
    tx.begin();
    entityManager.flush(); // 잠재적 위험
    entityManager.clear();
} // end if

• JpaPagingItemReaderBuilder.transacted(false) 설정으로 해당 설정을 끌 수 있다.
  • 다만 설정이 꺼지면 엔티티는 영속성 컨텍스트에서 분리되기에 (detach) ItemPrcoessor에서의 lazy loading이 불가능해진다
  • 사실 배치 처리에선 FetchType.EAGER가 더 적절할 때가 많다.

JpaItemWriter

• jpa 의존성을 추가하면 Spring boot를 사용하는 환경에선 자동으로 PlatformTransactionManager의 구현체가 JpaTransactionManager로 변경된다.
  • 영속성 컨텍스트, 1차 캐시 등의 jpa 특화 기능을 활용하는 구현체
  • 덕분에 JpaItemWriter가 동작할 수 있게 된다.
• JpaItemWriter는 단순하다.
  • 넘겨 받은 엔티티를 DB로 보내는 역할
  • 따로 쿼리를 작성할 필요도 없다.
  • 단 처리 대상이 jpa 엔티티여야 한다.

@Bean
fun postBlockWriter(): JpaItemWriter<BlockedPost> {
    return JpaItemWriterBuilder<BlockedPost>()
        .entityManagerFactory(entityManagerFactory)
        .usePersist(true) // persist를 하도록 설정
        .build()
}

• JpaItemWriterBuilder로 간단하게 구성할 수 있다.
  • 새로운 데이터를 추가한다면 persist
  • 기존 데이터를 수정한다면 merge를 사용하면 된다. (userPsersist(false), false가 기본값이다.)

IDENTITY 전략 사용 시 배치 처리 제약 사항

• JPA의 IDENTITY ID 생성 전략을 사용할 경우 INSERT 쿼리의 배치 처리를 할 수 없다.
  • IDENTITY 전략에선 엔티티 ID가 DB에서 생성되기에 엔티티 영속화를 위해선 반드시 INSERT를 먼저 실행해야 하기 때문
  • 즉 모든 INSERT가 개별적으로 실행되어야 함을 의미한다.
  • 배치 처리 성능이 중요하다면 SEQEUNCE 전략을 권장한다.