[SpringBatch 연재 09] 입맛에 맞는 배치 처리를 위한 Custom ItemReader/ItemWriter 구현방법 알아보기

KIDO 24.11.25

2,350 5 7

facebook twitter kakao link

5 7

facebook twitter kakao link

DEVOTEE 요약: 이번 블로그에서는 Spring Batch에서 커스텀 ItemReader와 ItemWriter를 구현하는 방법에 대해 다뤘습니다. 기본 제공되는 ItemReader와 ItemWriter로 충분하지 않은 경우, 비즈니스 로직에 맞게 Querydsl과 같은 도구를 사용하여 새로운 ItemReader인 QuerydslPagingItemReader를 만듭니다. 또한 Spring Batch의 기본 ItemWriter 인터페이스를 구현하여, 외부 서비스를 호출하는 CustomItemWriter를 작성하여 배치를 구성하였습니다. 이러한 맞춤형 구현은 확장성과 유연성을 제공하지만, 개발의 복잡성이 증가할 수 있습니다.

DEVOTEE 추천 블로그

이번시간에는 Custom ItemReader와 ItemWriter에 대해서 알아볼 것이다.

스프링 배치는 사전에 구현된 내장 ItemReader와 ItemWriter를 가지고 있어서 일반적인 케이스는 모두 수용이 가능하다.

그러나 특이 케이스나, 비즈니스 로직에 딱 맞는 배치를 수행하기 위해서 Customize가 필요하다.

- 우선 QueryDsl을 이용하여 Paging하여 데이터베이스의 값을 읽어 들이는 QuerydslPagingItemReader를 구현해 볼 것이다.

- 그리고 CustomItemWriter를 구현하여 타 서비스를 호출하는 샘플을 만들어 보자.

QuerydslPagingItemReader 개요

Querydsl은 SpringBatch의 공식 ItemReader이 아니다.
우리는 AbstractPagingItemReader를 이용하여 Querydsl 을 활용할 수 있도록 ItemReader를 만들 것이다.
Querydsl 기능 활용: Querydsl의 강력하고 유연한 쿼리 기능을 사용하여 데이터를 효율적으로 읽을 수 있다.
JPA 엔티티 추상화: JPA 엔티티에 직접 의존하지 않고 추상화된 쿼리를 작성하여 코드 유지 관리성을 높일 수 있다.
동적 쿼리 지원: 런타임 시 조건에 따라 동적으로 쿼리를 생성할 수 있다.

QuerydslPagingItemReader 생성하기

package com.schooldevops.springbatch.batchsample.jobs.jpa.reader;

import com.querydsl.jpa.impl.JPAQuery;
import com.querydsl.jpa.impl.JPAQueryFactory;
import jakarta.persistence.EntityManager;
import jakarta.persistence.EntityManagerFactory;
import org.springframework.batch.item.database.AbstractPagingItemReader;
import org.springframework.util.ClassUtils;
import org.springframework.util.CollectionUtils;

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

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

    private EntityManager em;
    private final Function<JPAQueryFactory, JPAQuery<T>> querySupplier;

    private final Boolean alwaysReadFromZero;

    public QuerydslPagingItemReader(EntityManagerFactory entityManagerFactory, Function<JPAQueryFactory, JPAQuery<T>> querySupplier, int chunkSize) {
        this(ClassUtils.getShortName(QuerydslPagingItemReader.class), entityManagerFactory, querySupplier, chunkSize, false);
    }

    public QuerydslPagingItemReader(String name, EntityManagerFactory entityManagerFactory, Function<JPAQueryFactory, JPAQuery<T>> querySupplier, int chunkSize, Boolean alwaysReadFromZero) {
        super.setPageSize(chunkSize);
        setName(name);
        this.querySupplier = querySupplier;
        this.em = entityManagerFactory.createEntityManager();
        this.alwaysReadFromZero = alwaysReadFromZero;

    }

    @Override
    protected void doClose() throws Exception {
        if (em != null)
            em.close();
        super.doClose();
    }

    @Override
    protected void doReadPage() {
        initQueryResult();

        JPAQueryFactory jpaQueryFactory = new JPAQueryFactory(em);
        long offset = 0;
        if (!alwaysReadFromZero) {
            offset = (long) getPage() * getPageSize();
        }

        JPAQuery<T> query = querySupplier.apply(jpaQueryFactory).offset(offset).limit(getPageSize());

        List<T> queryResult = query.fetch();
        for (T entity: queryResult) {
            em.detach(entity);
            results.add(entity);
        }
    }

    private void initQueryResult() {
        if (CollectionUtils.isEmpty(results)) {
            results = new CopyOnWriteArrayList<>();
        } else {
            results.clear();
        }
    }
}

public class QuerydslPagingItemReader extends AbstractPagingItemReader:
- 위 선언은 AbstractPagingItemReader 를 상속받았다.
- AbstractPagingItemReader은 어댑터 패턴으로, 상속받는 쪽은 doReadPage만 구현하면 된다.
생성자
```
    public QuerydslPagingItemReader(String name, EntityManagerFactory entityManagerFactory, Function<JPAQueryFactory, JPAQuery<T>> querySupplier, int chunkSize) {
        super.setPageSize(chunkSize);
        setName(name);
        this.querySupplier = querySupplier;
        this.em = entityManagerFactory.createEntityManager();

    }
```
- 위 내용과 같이 우리에게 필요한 부분은 name, entityManagerFactory, querySupplier, chunkSize 이다.
  - name: ItemReader를 구분하기 위한 이름이다.
  - entityManagerFactory: JPA를 이용하기 위해서 entityManagerFactory 를 전달한다.
  - Function<JPAQueryFactory, JPAQuery>: 이는 JPAQuery를 생성하기 위한 Functional Interface 이다.
    - 입력 파라미터로 JPAQueryFactory 를 입력으로 전달 받는다.
    - 반환값은 JPAQuery 형태의 queryDSL 쿼리가 된다.
  - chunkSize: 한번에 페이징 처리할 페이지 크기이다.
  - alwaysReadFromZero: 항상 0부터 페이징을 읽을지 여부를 지정한다. 만약 paging 처리된 데이터 자체를 수정하는경우 배치처리 누락이 발생할 수 있으므로 이를 해결하기 위한 방안으로 사용된다.
doClose()
- doClose는 기본적으로 AbstractPagingItemReader를 자체 구현되어 있지만 EntityManager자원을 해제하기 위해서 em.close() 를 수행한다.
doReadPage()
- 실제로 우리가 구현해야할 추상 메소드이다.
```
  @Override
  protected void doReadPage() {
      initQueryResult();

      JPAQueryFactory jpaQueryFactory = new JPAQueryFactory(em);
      long offset = 0;
      if (!alwaysReadFromZero) {
          offset = (long) getPage() * getPageSize();
      }

      JPAQuery<T> query = querySupplier.apply(jpaQueryFactory).offset(offset).limit(getPageSize());

      List<T> queryResult = query.fetch();
      for (T entity: queryResult) {
          em.detach(entity);
          results.add(entity);
      }
  }
```
- JPAQueryFactory를 통해서 함수형 인터페이스로 지정된 queryDSL에 적용할 QueryFactory이다.
- 만약 alwaysReadFromZero 가 false라면 offset과 limit을 계속 이동하면서 조회하도록 offset을 계산한다.
- querySupplier.apply
  - 우리가 제공한 querySupplier에 JPAQueryFactory를 적용하여 JPAQuery를 생성하도록한다.
  - 페이징을 위해서 offset, limit을 계산된 offset과 pageSize (청크크기) 를 지정하여 페이징 처리하도록 한다.
- fetch:
  - 결과를 패치하여 패치된 내역을 result에 담는다.
  - 이때 entityManager에서 detch하여 변경이 실제 DB에 반영되지 않도록 영속성 객체에서 제외시킨다.
initQueryResult
- 매 페이징 결과를 반환할때 페이징 결과만 반환하기 위해서 초기화한다.
- 만약 결과객체가 초기화 되어 있지 않다면 CopyOnWriteArrayList 객체를 신규로 생성한다.

편의를 위해서 Builder 생성하기

위 생성자는 조금 복잡했다. 이를 편하게 작성하기 위해서 빌더를 생성해보자.
QuerydslPagingItemReaderBuilder.java를 만들고 다음과 같이 작성한다.

package com.schooldevops.springbatch.batchsample.jobs.jpa.reader;

import com.querydsl.jpa.impl.JPAQuery;
import com.querydsl.jpa.impl.JPAQueryFactory;
import jakarta.persistence.EntityManager;
import jakarta.persistence.EntityManagerFactory;
import org.springframework.batch.item.database.AbstractPagingItemReader;
import org.springframework.util.ClassUtils;
import org.springframework.util.CollectionUtils;

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

public class QuerydslPagingItemReaderBuilder<T> {

    private EntityManagerFactory entityManagerFactory;
    private Function<JPAQueryFactory, JPAQuery<T>> querySupplier;

    private int chunkSize = 10;

    private String name;

    private Boolean alwaysReadFromZero;

    public QuerydslPagingItemReaderBuilder<T> entityManagerFactory(EntityManagerFactory entityManagerFactory) {
        this.entityManagerFactory = entityManagerFactory;
        return this;
    }

    public QuerydslPagingItemReaderBuilder<T> querySupplier(Function<JPAQueryFactory, JPAQuery<T>> querySupplier) {
        this.querySupplier = querySupplier;
        return this;
    }

    public QuerydslPagingItemReaderBuilder<T> chunkSize(int chunkSize) {
        this.chunkSize = chunkSize;
        return this;
    }

    public QuerydslPagingItemReaderBuilder<T> name(String name) {
        this.name = name;
        return this;
    }

    public QuerydslPagingItemReaderBuilder<T> alwaysReadFromZero(Boolean alwaysReadFromZero) {
        this.alwaysReadFromZero = alwaysReadFromZero;
        return this;
    }

    public QuerydslPagingItemReader<T> build() {
        if (name == null) {
            this.name = ClassUtils.getShortName(QuerydslPagingItemReader.class);
        }
        if (this.entityManagerFactory == null) {
            throw new IllegalArgumentException("EntityManagerFactory can not be null.!");
        }
        if (this.querySupplier == null) {
            throw new IllegalArgumentException("Function<JPAQueryFactory, JPAQuery<T>> can not be null.!");
        }
        if (this.alwaysReadFromZero == null) {
            alwaysReadFromZero = false;
        }
        return new QuerydslPagingItemReader<>(this.name, entityManagerFactory, querySupplier, chunkSize, alwaysReadFromZero);
    }
}

위 내용은 Builder 패턴을 이용하여 단순히 QuerydslPgingItemReader 객체를 생성하는 코드일 뿐이다.

소스 샘플

이제 전체 샘플을 이용하여 배치를 수행해보자.

package com.schooldevops.springbatch.batchsample.jobs.jpa;

import com.schooldevops.springbatch.batchsample.jobs.jpa.processor.CustomerItemProcessor;
import com.schooldevops.springbatch.batchsample.jobs.jpa.reader.QuerydslPagingItemReader;
import com.schooldevops.springbatch.batchsample.jobs.jpa.reader.QuerydslPagingItemReaderBuilder;
import com.schooldevops.springbatch.batchsample.jobs.models.Customer;
import com.schooldevops.springbatch.batchsample.jobs.models.QCustomer;
import jakarta.persistence.EntityManagerFactory;
import lombok.extern.slf4j.Slf4j;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.job.builder.JobBuilder;
import org.springframework.batch.core.launch.support.RunIdIncrementer;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.step.builder.StepBuilder;
import org.springframework.batch.item.file.FlatFileItemWriter;
import org.springframework.batch.item.file.builder.FlatFileItemWriterBuilder;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.io.FileSystemResource;
import org.springframework.transaction.PlatformTransactionManager;

import javax.sql.DataSource;

@Slf4j
@Configuration
public class QueryDSLPagingReaderJobConfig {

    /**
     * CHUNK 크기를 지정한다.
     */
    public static final int CHUNK_SIZE = 2;
    public static final String ENCODING = "UTF-8";
    public static final String QUERYDSL_PAGING_CHUNK_JOB = "QUERYDSL_PAGING_CHUNK_JOB";

    @Autowired
    DataSource dataSource;

    @Autowired
    EntityManagerFactory entityManagerFactory;

//    @Bean
//    public QuerydslPagingItemReader<Customer> customerQuerydslPagingItemReader() throws Exception {
//
//        Function<JPAQueryFactory, JPAQuery<Customer>> query = jpaQueryFactory -> jpaQueryFactory.select(QCustomer.customer).from(QCustomer.customer);
//
//        return new QuerydslPagingItemReader<>("customerQuerydslPagingItemReader", entityManagerFactory, query, CHUNK_SIZE, false);
//    }

    @Bean
    public QuerydslPagingItemReader<Customer> customerQuerydslPagingItemReader() {
        return new QuerydslPagingItemReaderBuilder<Customer>()
                .name("customerQuerydslPagingItemReader")
                .entityManagerFactory(entityManagerFactory)
                .chunkSize(2)
                .querySupplier(jpaQueryFactory -> jpaQueryFactory.select(QCustomer.customer).from(QCustomer.customer).where(QCustomer.customer.age.gt(20)))
                .build();
    }

    @Bean
    public FlatFileItemWriter<Customer> customerQuerydslFlatFileItemWriter() {

        return new FlatFileItemWriterBuilder<Customer>()
                .name("customerQuerydslFlatFileItemWriter")
                .resource(new FileSystemResource("./output/customer_new_v2.csv"))
                .encoding(ENCODING)
                .delimited().delimiter("\t")
                .names("Name", "Age", "Gender")
                .build();
    }


    @Bean
    public Step customerQuerydslPagingStep(JobRepository jobRepository, PlatformTransactionManager transactionManager) throws Exception {
        log.info("------------------ Init customerQuerydslPagingStep -----------------");

        return new StepBuilder("customerJpaPagingStep", jobRepository)
                .<Customer, Customer>chunk(CHUNK_SIZE, transactionManager)
                .reader(customerQuerydslPagingItemReader())
                .processor(new CustomerItemProcessor())
                .writer(customerQuerydslFlatFileItemWriter())
                .build();
    }

    @Bean
    public Job customerJpaPagingJob(Step customerJdbcPagingStep, JobRepository jobRepository) {
        log.info("------------------ Init customerJpaPagingJob -----------------");
        return new JobBuilder(QUERYDSL_PAGING_CHUNK_JOB, jobRepository)
                .incrementer(new RunIdIncrementer())
                .start(customerJdbcPagingStep)
                .build();
    }
}

우리는 이제 queryDSL을 이용하여 배치를 수행할 수 있게 되었다.
queryDSL은 공식 Spring Batch ItemReader이 아니므로 AbastractPagingItemReader를 상속받아 구현했다.
일반적인 queryDSL 페이징은 쉽게 구현할 수 있으며, 필요하다면 doReadPage()의 내용을 조금 수정하면 원하는 방향으로 개발할 수 있을 것이다.

CustomItemWriter 개요

CustomItemWriter는 Spring Batch에서 제공하는 기본 ItemWriter 인터페이스를 구현하여 직접 작성한 ItemWriter 클래스이다.
기본 ItemWriter 클래스로는 제공되지 않는 특정 기능을 구현할 때 사용된다

구성 요소

ItemWriter 인터페이스 구현: write() 메소드를 구현하여 원하는 처리를 수행한다.
필요한 라이브러리 및 객체 선언: 사용할 라이브러리 및 객체를 선언한다.
데이터 처리 로직 구현: write() 메소드에서 데이터 처리 로직을 구현한다.

장점

유연성: 기본 ItemWriter 클래스로는 제공되지 않는 특정 기능을 구현할 수 있다.
확장성: 다양한 방식으로 데이터 처리를 확장할 수 있다.
제어 가능성: 데이터 처리 과정을 완벽하게 제어할 수 있다.

단점

개발 복잡성: 기본 ItemWriter 클래스보다 개발 과정이 더 복잡하다.
테스트 어려움: 테스트 작성이 더 어려울 수 있다.
디버깅 어려움: 문제 발생 시 디버깅이 더 어려울 수 있다.

소스 샘플

다음은 커스텀 Writer를 작성했다.
개념을 간단히 설명하기 위해 최대한 간단한 예제를 작성하였다.

CustomService

청크 아이템을 받아서 호출하는 서비스를 작성한다.

package com.schooldevops.springbatch.batchsample.jobs.mybatiswriter;

import com.schooldevops.springbatch.batchsample.jobs.models.Customer;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import org.springframework.stereotype.Service;

import java.util.Map;

@Slf4j
@Service
public class CustomService {

    public Map<String, String> processToOtherService(Customer item) {

        log.info("Call API to OtherService....");

        return Map.of("code", "200", "message", "OK");
    }
}

우리가 알고 있는 일반적인 서비스 객체를 하나 만들었고, 단순히 log정보를 출력하고 Map결과를 반환하는 역할이다.

CustomItemWriter 작성하기

ItemWriter 인터페이스를 구현한 CustomItemWriter.java 파일에 다음과 같이 작성하자.

package com.schooldevops.springbatch.batchsample.jobs.mybatiswriter;

import com.schooldevops.springbatch.batchsample.jobs.models.Customer;
import lombok.extern.slf4j.Slf4j;
import org.springframework.batch.item.Chunk;
import org.springframework.batch.item.ItemWriter;
import org.springframework.stereotype.Component;

@Slf4j
@Component
public class CustomItemWriter implements ItemWriter<Customer> {

    private final CustomService customService;

    public CustomItemWriter(CustomService customService) {
        this.customService = customService;
    }

    @Override
    public void write(Chunk<? extends Customer> chunk) throws Exception {
        for (Customer customer: chunk) {
            log.info("Call Porcess in CustomItemWriter...");
            customService.processToOtherService(customer);
        }
    }
}

이전에 만든 CustomService를 생성자 파라미터로 받고, write 메소드를 구현하였다.
write메소드는 우리가 작성할 ItemWriter의 핵심 메소드이다.
Chunk는 Customer 객체를 한묶음으로 처리할수 있도록 반복 수행할 수 있다. 하여 for 구분을 사용했다.
이 메소드 내부에서 processToOtherSerive를 호출하였다.

전체코드

package com.schooldevops.springbatch.batchsample.jobs.mybatiswriter;


import com.schooldevops.springbatch.batchsample.jobs.models.Customer;
import lombok.extern.slf4j.Slf4j;
import org.apache.ibatis.session.SqlSessionFactory;
import org.mybatis.spring.batch.MyBatisBatchItemWriter;
import org.mybatis.spring.batch.builder.MyBatisBatchItemWriterBuilder;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.job.builder.JobBuilder;
import org.springframework.batch.core.launch.support.RunIdIncrementer;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.step.builder.StepBuilder;
import org.springframework.batch.item.file.FlatFileItemReader;
import org.springframework.batch.item.file.builder.FlatFileItemReaderBuilder;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.io.ClassPathResource;
import org.springframework.transaction.PlatformTransactionManager;

import javax.sql.DataSource;
import java.util.HashMap;
import java.util.Map;

@Slf4j
@Configuration
public class MybatisItemWriterJobConfig {

    /**
     * CHUNK 크기를 지정한다.
     */
    public static final int CHUNK_SIZE = 100;
    public static final String ENCODING = "UTF-8";
    public static final String MY_BATIS_ITEM_WRITER = "MY_BATIS_ITEM_WRITER";

    @Autowired
    DataSource dataSource;

    @Autowired
    SqlSessionFactory sqlSessionFactory;

    @Autowired
    CustomItemWriter customItemWriter;

    @Bean
    public FlatFileItemReader<Customer> flatFileItemReader() {

        return new FlatFileItemReaderBuilder<Customer>()
                .name("FlatFileItemReader")
                .resource(new ClassPathResource("./customer.csv"))
                .encoding(ENCODING)
                .delimited().delimiter(",")
                .names("name", "age", "gender")
                .targetType(Customer.class)
                .build();
    }

    @Bean
    public Step flatFileStep(JobRepository jobRepository, PlatformTransactionManager transactionManager) {
        log.info("------------------ Init flatFileStep -----------------");

        return new StepBuilder("flatFileStep", jobRepository)
                .<Customer, Customer>chunk(CHUNK_SIZE, transactionManager)
                .reader(flatFileItemReader())
                .writer(customItemWriter)
                .build();
    }

    @Bean
    public Job flatFileJob(Step flatFileStep, JobRepository jobRepository) {
        log.info("------------------ Init flatFileJob -----------------");
        return new JobBuilder(MY_BATIS_ITEM_WRITER, jobRepository)
                .incrementer(new RunIdIncrementer())
                .start(flatFileStep)
                .build();
    }
}

이제 우리는 customItemWriter 를 Autowired 하고, 단순하게 writer에 할당하면 된다.

결과보기

2024-04-11T13:40:08.142+09:00  INFO 57064 --- [           main] o.s.batch.core.job.SimpleStepHandler     : Executing step: [flatFileStep]
2024-04-11T13:40:08.163+09:00  INFO 57064 --- [           main] c.s.s.b.j.m.CustomItemWriter             : Call Porcess in CustomItemWriter...
2024-04-11T13:40:08.163+09:00  INFO 57064 --- [           main] c.s.s.b.j.mybatiswriter.CustomService    : Call API to OtherService....
2024-04-11T13:40:08.163+09:00  INFO 57064 --- [           main] c.s.s.b.j.m.CustomItemWriter             : Call Porcess in CustomItemWriter...
2024-04-11T13:40:08.163+09:00  INFO 57064 --- [           main] c.s.s.b.j.mybatiswriter.CustomService    : Call API to OtherService....
2024-04-11T13:40:08.163+09:00  INFO 57064 --- [           main] c.s.s.b.j.m.CustomItemWriter             : Call Porcess in CustomItemWriter...
2024-04-11T13:40:08.163+09:00  INFO 57064 --- [           main] c.s.s.b.j.mybatiswriter.CustomService    : Call API to OtherService....