Spring Batch ItemReader 성능 개선, 근데 이제 Querydsl을 곁들인...

 

develetter

https://github.com/prgrms-be-devcourse/NBE1_2_Team07

GitHub - prgrms-be-devcourse/NBE1_2_Team07: programmers devCourse BE 1기 7팀 2차 프로젝트

 

현재 개발 중인 서비스, develetter는 개발자 취준생들에게 맞춤형 뉴스레터로 유용한 정보를 제공하는 서비스입니다.

 

 

develetter는 두 가지 주요 API를 활용하여 데이터를 수집하고 있습니다.

사람인 채용 정보 API -> 채용 정보

Google Programmable Search Engine API -> 기술 블로그, IT 컨퍼런스

 

1. 데이터는 API를 호출하여 갱신됩니다.
2. 유저가 설정해 둔 관심 키워드로 데이터를 필터링합니다. (Spring Batch 적용)
3. 이후 필터링된 결과를 미리 만들어 둔 메일 템플릿에 삽입하여, 정해진 시간에 맞춰 뉴스레터 형식으로 메일을 발송합니다.

위 모든 작업은 스케줄러로 관리됩니다.

---

 

Spring Batch ItemReader

 

 

사실, 이번 포스팅에서 중점적으로 다룰 내용은 제목에서도 알 수 있지만 스프링 배치(Spring Batch)의 성능 개선입니다.

 

조금 더 자세히 말하자면, 사람인 채용정보 API로 수집된 DB의 채용 정보를 Batch Job의 Reader 단계에서 처리하는 성능을 최적화하는 과정에 초점을 맞췄습니다.

 

기존에 사용했던 RepositoryItemReader보다 더 효율적으로 데이터를 읽어오는 방법을 모색한 결과, Querydsl을 활용한 성능 개선을 이뤄냈습니다.

 

Querydsl을 사용함으로 컴파일시 쿼리를 검증할 수 있는 일석이조의 효과를 누릴 수 있었지만, 문제는 스프링 배치는 Querydsl을 공식적으로 지원하지 않아서 직접 구현해야한다는 점입니다.

 

 

 

이제, Querydsl을 통해서 Batch Job의 Reader 성능을 개선하는 과정과 그 결과에 대해 설명해보겠습니다.

---

 

RepositoryItemReader

 

    @Bean
    public RepositoryItemReader<JobPosting> reader() {
        return new RepositoryItemReaderBuilder<JobPosting>()
                .name("jobPostingItemReader")
                .pageSize(chunkSize)
                .repository(jobPostingRepository)
                .methodName("findAll") // 모든 JobPosting을 조회
                .sorts(Map.of("id", Sort.Direction.ASC))
                .build();
    }

 

위 코드는 저희 팀에서 사용한 기존의 ItemReader 입니다.

기본적으로 Spring Batch에서 제공하는 ItemReader들 중에서 JPQL이나 SQL 없이, 사용법이 가장 깔끔해보여서 선택했었습니다!

 

하지만 생각보다 처참한 속도로 무엇인가 잘못된 것이라는 것을 쉽게 인지할 수 있었고 Spring Batch 성능 개선에 대해서 구글링했습니다.

 

사실, 이때까지는 RepositoryItemReader가 문제라고 생각하지 못했습니다...

---

 

QuerydslPagingItemReader

 

 

QuerydslPagingItemReader 사용 예시입니다!

쿼리를 람다로 표현할 수 있게 만들어둬서 하나의 Reader 클래스로 여러 조건을 변경하며 사용할 수 있습니다. 

    @Bean
    public QuerydslPagingItemReader<JobPosting> reader() {
        return new QuerydslPagingItemReader<>(emf, chunkSize, 
        queryFactory -> queryFactory.selectFrom(QJobPosting.jobPosting));
    }

 

 

 

조금은 어지러운, 하지만 알고 가면 좋은 ItemReader 클래스들의 전체적인 구조.

역시 열어주셨군요...

AbstractPagingItemReader의 abstract method doReadPage

 

AbstractPagingItemReader 클래스, abstract method doReadPage가 보입니다. createQuery는 보이지 않네요.

 

 

 

아래 코드에서, AbstractPagingItemReader를 상속받은 JpaPagingItemReader를 확인해보면, createQuery와 그를 호출하는 doReadPage를 확인할 수 있습니다. 

 

저희의 목표는 저 동작을 Querydsl로 변경하는 것이고, private로 선언된 createQuery 메서드때문에 JpaPagingItemReader가 아닌 AbstractPagingItemReader를 상속받는 것입니다.

JpaPagingItemReader의 createQuery, doReadPage

 

아래 코드에서 저희가 구현한 QuerydslPagingItemReader가 AbstractPagingItemReader 클래스를 상속받고 있음을 확인할 수 있습니다.

 

상속의 주된 목표는, AbstractPagingItemReader에서 제공하는 추상 메서드인 doReadPage()를 구현하고, 그 내부에서 createQuery() 메서드를 호출하는 것입니다.

package com.develetter.develetter.jobposting.batch;

import com.develetter.develetter.jobposting.entity.JobPosting;
import com.develetter.develetter.jobposting.entity.QJobPosting;
import com.querydsl.jpa.JPQLQuery;
import com.querydsl.jpa.impl.JPAQuery;
import com.querydsl.jpa.impl.JPAQueryFactory;
import jakarta.persistence.EntityManager;
import jakarta.persistence.EntityManagerFactory;
import jakarta.persistence.EntityTransaction;
import lombok.Setter;
import org.springframework.batch.item.database.AbstractPagingItemReader;
import org.springframework.dao.DataAccessResourceFailureException;
import org.springframework.util.ClassUtils;
import org.springframework.util.CollectionUtils;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.function.Function;

public class QuerydslPagingItemReader<T> extends AbstractPagingItemReader<T> {

    protected final Map<String, Object> jpaPropertyMap = new HashMap<>();
    protected EntityManagerFactory entityManagerFactory;
    protected EntityManager entityManager;
    protected Function<JPAQueryFactory, JPAQuery<T>> queryFunction;
    @Setter
    protected boolean transacted = false; // default value = true
    private Long lastId;

    protected QuerydslPagingItemReader() {
        setName(ClassUtils.getShortName(QuerydslPagingItemReader.class));
    }

    public QuerydslPagingItemReader(EntityManagerFactory entityManagerFactory,
                                    int pageSize,
                                    Function<JPAQueryFactory, JPAQuery<T>> queryFunction) {
        this(entityManagerFactory, pageSize, true, queryFunction);
    }

    public QuerydslPagingItemReader(EntityManagerFactory entityManagerFactory,
                                    int pageSize,
                                    boolean transacted,
                                    Function<JPAQueryFactory, JPAQuery<T>> queryFunction) {
        this();
        this.entityManagerFactory = entityManagerFactory;
        this.queryFunction = queryFunction;
        setPageSize(pageSize);
        setTransacted(transacted);
    }


    @Override
    protected void doOpen() throws Exception {
        super.doOpen();

        entityManager = entityManagerFactory.createEntityManager(jpaPropertyMap);
        if (entityManager == null) {
            throw new DataAccessResourceFailureException("Unable to obtain an EntityManager");
        }
    }

    @Override
    protected void doReadPage() {
        EntityTransaction tx = getTxOrNull();

        JPQLQuery<T> query = createQuery()
                .where(lastId != null ? QJobPosting.jobPosting.id.gt(lastId) : null)
                .limit(getPageSize());

        initResults();

        fetchQuery(query, tx);

        if (!results.isEmpty()) {
            T lastEntity = results.get(results.size() - 1);
            lastId = extractIdFromEntity(lastEntity);
        }
    }

    protected Long extractIdFromEntity(T entity) {
        return ((JobPosting) entity).getId();
    }

    protected EntityTransaction getTxOrNull() {
        if (transacted) {
            EntityTransaction tx = entityManager.getTransaction();
            tx.begin();

            entityManager.flush();
            entityManager.clear();
            return tx;
        }

        return null;
    }

    protected JPAQuery<T> createQuery() {
        JPAQueryFactory queryFactory = new JPAQueryFactory(entityManager);
        return queryFunction.apply(queryFactory);
    }

    protected void initResults() {
        if (CollectionUtils.isEmpty(results)) {
            results = new CopyOnWriteArrayList<>();
        } else {
            results.clear();
        }
    }
    
    protected void fetchQuery(JPQLQuery<T> query, EntityTransaction tx) {
        if (transacted) {
            results.addAll(query.fetch());
            if(tx != null) {
                tx.commit();
            }
        } else {
            List<T> queryResult = query.fetch();
            for (T entity : queryResult) {
                entityManager.detach(entity);
                results.add(entity);
            }
        }
    }

    @Override
    protected void jumpToItem(int itemIndex) {
    }

    @Override
    protected void doClose() throws Exception {
        entityManager.close();
        super.doClose();
    }
}

 

엔티티 매니저 팩토리와 Querydsl 람다식, 페이지 크기만 설정하면 작동하는 QuerydslPagingItemReader가 완성됐습니다!

 

참고로 쿼리의 페이지 크기는 배치의 청크 크기와 맞춰야한다고 합니다! 그게 가장 효율적이고 일반적이기에...

 

 

성능 개선 관점에서도 살펴보도록 하겠습니다.

---

 

첫번째로 주목해야할 점은 doReadPage 에서 쿼리를 만드는 과정입니다.

    @Override
    protected void doReadPage() {
        EntityTransaction tx = getTxOrNull();

        JPQLQuery<T> query = createQuery()
                .where(lastId != null ? QJobPosting.jobPosting.id.gt(lastId) : null)
                .limit(getPageSize());

        initResults();

        fetchQuery(query, tx);

        if (!results.isEmpty()) {
            T lastEntity = results.get(results.size() - 1);
            lastId = extractIdFromEntity(lastEntity);
        }
    }

 

쿼리를 만들 때, 명시적인 Offset 설정이 없습니다.

대신, 조건(where)에서 현재까지 읽은 데이터의 마지막 Id(pk)를 시작으로 pageSize만큼 가져오고 있습니다!

 

이러한 방법을 통해서 페이징을 포함하는 select를 최적화할 수 있습니다. 

자세한 설명은 아래와 같습니다.

case 1) Offset을 사용한다면 100개를 pageSize = 10으로 조회할 때, 아래와 같이 조회합니다.

 1. 1 ~ 10  조회,    1 ~ 10 사용
 2. 1 ~ 20 조회,  11 ~ 20 사용
 3. 1 ~ 30 조회,  21 ~ 30 사용
...
10. 1 ~ 100 조회, 91 ~ 100 사용

(10 + 20 + 30 + ... + 100)건을 10번의 조회로 처리합니다.

 

case 2) Offset을 사용하지 않고, where, lastId를 사용해서 100개를 pageSize = 10으로 조회할 때, 아래와 같이 조회합니다.

 1.  1 ~ 10  조회,    1 ~ 10 사용
 2. 11 ~ 20 조회,  11 ~ 20 사용
 3. 21 ~ 30 조회,  21 ~ 30 사용
...
10. 91 ~ 100 조회, 91 ~ 100 사용

100건을 10번의 조회로 처리합니다.

즉, Offset 설정이 아닌, Id(pk)를 활용한 where절을 사용함으로 read 과정을 최적화할 수 있습니다.

 

이외에도 커버링 인덱스를 사용하는 방법도 있다고 합니다. 저는 where 와 lastId(pk)를 사용하는 방법만으로 충분했습니다.

 

---

 

두번째로 주목해야할 점은 transacted 멤버 변수를 false로 설정한 것입니다.

    protected boolean transacted = false; // default value = true

 

더 넓은 범위의 청크(Chunk)에서 이미 트랜잭션을 관리해주기 때문에, Reader 내부에서의 트랜잭션 관리는 불필요하다고 판단하여 기본적으로 활성화되어 있던 트랜잭션을 비활성화했습니다.

 

---

 

 

 

 

성능 비교

 

참고: 개선된 Reader뿐만 아니라 기존의 Processor, Writer가 포함된 작업의 결과값을 간단히 포스트맨으로 검증했습니다.

 

---

페이지, 청크 사이즈가 모두 10인 경우

 

기존 43.64s -----> 개선 2.46s

기존의 RepositoryItemReader

Querydsl(NoOffset)PagingItemReader

 

---

페이지, 청크 사이즈가 모두 100인 경우

 

기존 31.26s -----> 개선 1.692s

기존의 RepositoryItemReader

 

Querydsl(NoOffset)PagingItemReader

 

 

 

약 20배의 성능 향상이 이뤄짐을 확인할 수 있습니다.

현재 데이터셋은 배치를 적용할 수준의 큰 규모는 아니지만, 유의미한 결과를 도출할 수 있었습니다.

 

자료를 참고하면서도 시행착오가 많았지만, 또 해결해버렸습니다. 행복해요.

 

---

 

참고 자료

https://techblog.woowahan.com/2662/

Spring Batch와 Querydsl | 우아한형제들 기술블로그

https://github.com/jojoldu/spring-batch-querydsl

GitHub - jojoldu/spring-batch-querydsl: 스프링배치와 QuerydslPagingItemReader

 

 

https://tech.kakaopay.com/post/ifkakao2022-batch-performance-read/

[if kakao 2022] Batch Performance를 고려한 최선의 Reader | 카카오페이 기술 블로그

 

 

 

https://jaeseo0519.tistory.com/400

[Spring Boot] Batch 성능 개선기 (+`24.07.25 추가 개선 및 테스트)