[Spring Batch - 07] MyBatisPagingItemReader와 MyBatisBatchItemWriter

reference :

[SpringBatch 연재 07] MyBatisPagingItemReader로 DB내용을 읽고, MyBatisItemWriter로 DB에 쓰기

 

 

Review

지금까지 아래와 같은 Spring Batch의 ItemReader와 ItemWriter 구현체들을 살펴보았다.

• FlatFileItemReader와 FlatFileItemWriter
• JdbcPagingItemReader와 JdbcBatchItemWriter
• JpaPagingItemReader와 JpaItemWriter

 

이번 포스팅에서는 MyBatisPagingItemReader와 MyBatisBatchItemWriter에 대해 알아보자. 이들은 데이터베이스를 데이터 소스로 하고 MyBatis를 통해 SQL 매핑을 한다. 

MyBatis는 Java에서 데이터베이스 접근을 위한 프레임워크의 일종이다. SQL 쿼리를 XML 파일에 명시적으로 작성하고, 이를 이용해 데이터를 가져오거나 저장한다. SQL을 직접 작성하므로, 복잡한 쿼리를 더 효율적으로 관리할 수 있다는 장점이 있다. 반면 XML 설정이 많고 기본적인 데이터 매핑이 JPA 등의 다른 ORM 보다 번거롭다는 단점이 있다.

 

지금까지 공부한 ItemReader와 ItemWriter의 특징을 정리하면 아래와 같다.

	FlatFileItemReader/FlatFileItemWriter	JdbcPagingItemReader/JdbcBatchItemWriter	JpaPagingItemReader/JpaItemWriter	MyBatisPagingItemReader/MyBatisBatchItemWriter
데이터 소스	파일 (.csv, .txt 등)	데이터베이스 (JDBC)	데이터베이스 (JPA ORM)	데이터베이스 (MyBatis)
데이터 처리	파일에서 데이터를 읽어와 처리	SQL 쿼리를 직접 실행	ORM(Object-Relational Mapping)	MyBatis 매퍼로 SQL 실행
페이징	X	O	O	O
SQL 작성	X	O	X	O

 

 

ItemReader - MyBatisPagingItemReader

MyBatisPagingItemReader는 MyBatis를 이용해 데이터베이스에서 데이터를 페이지 단위로 읽어온다. 지정한 페이지 사이즈만큼씩 데이터를 나누어 가져오기 때문에 메모리 부담을 줄일 수 있다.

 

다음으로 MyBatisPagingItemReader를 사용한 예제를 살펴보기 전에 먼저 아래와 같은 설정이 필요하다.

 

[build.gradle]

dependencies 란에 아래와 같이 mybatis-spring-boot-starter를 추가해준다.

dependencies {
		implementation 'org.mybatis.spring.boot:mybatis-spring-boot-starter:3.0.1'
}

 

[application.yaml]

Spring Batch가 이번 예제에서 실행할 Job 이름을 정하고 추가해준다.

예제에서 사용할 .xml 파일의 경로도 추가해준다. 기본 classpath 경로는 /src/main/resources이다.

spring:
  batch:
    job:
      name: MYBATIS_CHUNK_JOB

mybatis:
  mapper-locations: classpath:/customer.xml

 

[customer.csv]

임의로 데이터를 지정하여  /src/main/resources 경로에 저장한다. 데이터는 name, age, gender 순으로 적는다.

alice,25,female
bob,34,Male
julia,29,female
michael,21,Male
emily,45,Female
chris,31,Male

 

[customer.xml]

MyBatis 매퍼로 사용될 customer.xml 파일을 아래와 같이 작성한다.

MyBatisPagingItemReader 예제에서는 selectCustomer 라는 식별자를 가진 SELECT 쿼리를 사용하여, customer 테이블에서 id, name, age, gender column을 조회할 것이다.

이때 MyBatis에서 제공하는 #{_skiprows}와 #{_pagesize} 페이징 변수를 동적으로 사용해 특정 범위의 row만 가져오도록 제한하였다.

<?xml version="1.0" encoding="UTF-8" ?>
<!DOCTYPE mapper
        PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
        "http://mybatis.org/dtd/mybatis-3-mapper.dtd">

<mapper namespace="com.spring_batch.batch_sample.jobs">

    <resultMap id="customerResult" type="com.spring_batch.batch_sample.jobs.models.Customer">
        <result property="id" column="id"/>
        <result property="name" column="name"/>
        <result property="age" column="age"/>
        <result property="gender" column="gender"/>
    </resultMap>

    <select id="selectCustomers" resultMap="customerResult">
        SELECT id, name, age, gender
        FROM customer
                 LIMIT #{_skiprows}, #{_pagesize}
    </select>

</mapper>

 

[Customer.java]

Customer.java 파일은 지난번에 작성한 것을 그대로 사용한다.

package com.spring_batch.batch_sample.jobs.models;

import lombok.Data;

@Data
public class Customer {
    private Long id;
    private String name;
    private int age;
    private String gender;
}

 

[CustomerItemProcessor.java]

CustomerItemProcessor.java 파일은 지난번에 작성한 것을 그대로 사용한다.

package com.spring_batch.batch_sample.jobs.mybatis;

import com.spring_batch.batch_sample.jobs.models.Customer;
import lombok.extern.slf4j.Slf4j;
import org.springframework.batch.item.ItemProcessor;

@Slf4j
public class CustomerItemProcessor implements ItemProcessor<Customer, Customer> {
    @Override
    public Customer process(Customer item) throws Exception {
        log.info("Item Processor ------------------- {}", item);
        return item;
    }
}

 

 

[MyBatisReaderConfig.java]

이제 MyBatisPagingItemReader를 사용하여 데이터베이스에서 데이터를 읽고 FlatFileItemWriter를 사용하여 CSV 파일로 저장하는 예제를 살펴보자.

package com.spring_batch.batch_sample.jobs.mybatis;

import com.spring_batch.batch_sample.jobs.models.Customer;
import lombok.extern.slf4j.Slf4j;
import org.apache.ibatis.session.SqlSessionFactory;
import org.mybatis.spring.batch.MyBatisPagingItemReader;
import org.mybatis.spring.batch.builder.MyBatisPagingItemReaderBuilder;
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 MyBatisReaderJobConfig {

    public static final int CHUNK_SIZE = 2; // CHUNK 크기 지정
    public static final String ENCODING = "UTF-8"; // 인코딩 형식 지정
    public static final String MYBATIS_CHUNK_JOB = "MYBATIS_CHUNK_JOB"; // Job 이름 지정

    @Autowired
    DataSource dataSource;

    @Autowired
    SqlSessionFactory sqlSessionFactory;

    @Bean
    public MyBatisPagingItemReader<Customer> myBatisItemReader() throws Exception {

        return new MyBatisPagingItemReaderBuilder<Customer>()
                .sqlSessionFactory(sqlSessionFactory)
                .pageSize(CHUNK_SIZE)
                .queryId("com.spring_batch.batch_sample.jobs.selectCustomers")
                .build();
    }


    @Bean
    public FlatFileItemWriter<Customer> customerCursorFlatFileItemWriter() {
        return new FlatFileItemWriterBuilder<Customer>()
                .name("customerCursorFlatFileItemWriter")
                .resource(new FileSystemResource("./output/customer_new_v4.csv"))
                .encoding(ENCODING)
                .delimited().delimiter(",")
                .names("Name", "Age", "Gender")
                .build();
    }


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

        return new StepBuilder("customerJdbcCursorStep", jobRepository)
                .<Customer, Customer>chunk(CHUNK_SIZE, transactionManager)
                .reader(myBatisItemReader())
                .processor(new CustomerItemProcessor())
                .writer(customerCursorFlatFileItemWriter())
                .build();
    }

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

1. MyBatisPagingItemReader :
   • sqlSessionFactory: MyBatis와 데이터베이스를 연결하는 팩토리를 지정한다. 이 객체를 통해 MyBatis가 데이터베이스에 접근한다.
   • pageSize: 한 번에 읽어올 데이터 크기를 지정한다.
   • queryId: 실행할 MyBatis SQL 쿼리의 ID를 지정한다. (customer.xml 파일에 작성한 SELECT 문)
2. FlatFileItemWriter: 지난 포스팅에서 살펴보았다.
3. Step: 
   • reader로 myBatisItemReader()를 지정한다.
   • processor로 new CustomerItemProcessor()를 지정한다.
   • writer로 customerCursorFlatFileItemWriter()를 지정한다.
4. Job:
   • JobBuilder: MYBATIS_CHUNK_JOB 이름의 Job을 생성한다.
   • incrementer(new RunIdIncrementer()): Job 실행 시 고유 ID를 생성하여 여러번 실행 가능하도록 한다.
   • start(customerJdbcCursorStep): Job의 첫 번째 Step으로 customerJdbcCursorStep을 실행한다.

 

[실행 결과]

 

 

ItemWriter - MyBatisBatchItemWriter

MyBatisBatchItemWriter는 MyBatis를 이용해 데이터베이스에 데이터를 배치 방식으로 저장한다. MyBatis 매퍼에 정의된 SQL 쿼리문을 실행하여 데이터를 일괄적으로 작성한다.

 

다음으로 MyBatisBatchItemWriter를 사용한 예제를 살펴보기 전에 먼저 아래와 같은 설정이 필요하다.

 

[build.gradle], [Customer.java]는 MyBatisPagingItemReader와 동일한 것을 사용한다.

 

[application.yaml]

이번 예제에서 실행할 Job 이름으로 변경해준다.

spring:
  batch:
    job:
      name: MY_BATIS_ITEM_WRITER

 

[customer.xml]

mapper 안에 MyBatisBatchItemWriter에서 사용할 insertCustomers라는 식별자를 가진 INSERT 쿼리를 추가한다. #{} 문법으로 Java 객체들의 값을 SQL에 동적 바인딩하여 데이터베이스의 customer2 테이블에 저장하는 쿼리이다. 

<mapper namespace="com.spring_batch.batch_sample.jobs">

    .
    .
    .
    
    <insert id="insertCustomers" parameterType="com.spring_batch.batch_sample.jobs.models.Customer">
        INSERT INTO customer2(name, age, gender) VALUES (#{name}, #{age}, #{gender});
    </insert>
</mapper>

 

[MyBatisWriterJobConfig]

이제 FlatFileItemReader를 사용하여 CSV 파일에서 데이터를 읽고, MyBatisBatchItemWriter를 사용하여 데이터베이스에 데이터를 저장하는 예제를 살펴보자.

package com.spring_batch.batch_sample.jobs.mybatis;

import com.spring_batch.batch_sample.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;

@Slf4j
@Configuration
public class MyBatisWriterJobConfig {
    /**
     * 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;

    @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 MyBatisBatchItemWriter<Customer> mybatisItemWriter() {
        return new MyBatisBatchItemWriterBuilder<Customer>()
                .sqlSessionFactory(sqlSessionFactory)
                .statementId("com.spring_batch.batch_sample.jobs.insertCustomers")
                .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(mybatisItemWriter())
                .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();
    }
}

1. FlatFileItemReader: 지난 포스팅에서 살펴보았다.
2. MyBatisBatchItemWriter :
   • sqlSessionFactory: MyBatis와 데이터베이스를 연결하는 팩토리를 지정한다.
   • statementId: 실행할 MyBatis SQL 쿼리의 ID를 지정한다. (customer.xml 파일에 작성한 INSERT 문)
3. Step:
   • reader로 flatFileItemReader()를 지정한다.
   • writer로 myBatisItemWriter()를 지정한다.
4. Job:
   • JobBuilder: MY_BATIS_ITEM_WRITER 이름의 Job을 생성한다.
   • incrementer(new RunIdIncrementer()): Job 실행 시 고유 ID를 생성하여 여러번 실행 가능하도록 한다.
   • start(flatFileStep): Job의 첫 번째 Step으로 flatFileStep을 실행한다.

 

[실행 결과]