[SpringBatch 연재 05] JdbcPagingItemReader로 DB내용을 읽고, JdbcBatchItemWriter로 DB에 쓰기

KIDO 24.10.28

2,548 7 1

facebook twitter kakao link

7 1

facebook twitter kakao link

DEVOTEE 요약: JdbcPagingItemReader는 Spring Batch에서 대규모 데이터를 효율적으로 처리하기 위해 데이터베이스로부터 페이지 단위로 데이터를 읽는 ItemReader입니다. 데이터베이스 커서를 사용해 페이지 크기, 커밋 간격, 쿼리 최적화를 설정할 수 있으며, 재시작 시 작업 상태를 저장하는 기능도 제공합니다. JdbcBatchItemWriter는 JDBC를 통해 데이터베이스에 데이터를 저장하는 ItemWriter로, 다양한 데이터베이스와 연동할 수 있으며, 설정 복잡성이 있으나 대량의 데이터를 빠르게 저장할 수 있는 이점이 있습니다.

JdbcPagingItemReader 개요

JdbcPagingItemReader는 Spring Batch에서 제공하는 ItemReader로, 데이터베이스로부터 데이터를 페이지 단위로 읽는다.
대규모 데이터 처리 효율성: 메모리 사용량을 최소화하고 커밋 간격을 설정하여 대규모 데이터를 효율적으로 처리할 수 있다.
쿼리 최적화: SQL 쿼리를 직접 작성하여 최적화된 데이터 읽기가 가능하다.
커서 제어: 데이터베이스 커서를 사용하여 데이터 순회를 제어할 수 있다.

JdbcPagingItemReader 주요 구성 요소

DataSource: 데이터베이스 연결 정보를 설정한다.
SqlQuery: 데이터를 읽을 SQL 쿼리를 설정한다.
RowMapper: SQL 쿼리 결과를 Item으로 변환하는 역할을 한다.
PageSize: 페이지 크기를 설정한다.
SkippableItemReader: 오류 발생 시 해당 Item을 건너뛸 수 있도록 한다.
ReadListener: 읽기 시작, 종료, 오류 발생 등의 이벤트를 처리할 수 있도록 한다.
SaveStateCallback: 잡 중단 시 현재 상태를 저장하여 재시작 시 이어서 처리할 수 있도록 한다.

JdbcPagingItemReader 샘플코드

Customer 클래스 생성

@Data
public class Customer {

    private String name;
    private int age;
    private String gender;

}

쿼리 Provider 생성하기

쿼리 Provider는 실제 배치를 위해서 데이터를 읽어올 쿼리를 작성한다.

    @Bean
    public PagingQueryProvider queryProvider() throws Exception {
        SqlPagingQueryProviderFactoryBean queryProvider = new SqlPagingQueryProviderFactoryBean();
        queryProvider.setDataSource(dataSource);  // DB 에 맞는 PagingQueryProvider 를 선택하기 위함
        queryProvider.setSelectClause("id, name, age, gender");
        queryProvider.setFromClause("from customer");
        queryProvider.setWhereClause("where age >= :age");

        Map<String, Order> sortKeys = new HashMap<>(1);
        sortKeys.put("id", Order.DESCENDING);

        queryProvider.setSortKeys(sortKeys);

        return queryProvider.getObject();
    }

SqlPagingQueryProviderFactoryBean: 쿼리 프로파이더 팩토리
setDataSource: 데이터소스를 설정한다.
setSelectClause: select에서 프로젝션할 필드 이름을 지정한다.
setFromClause: 조회할 테이블
setWhereClause: 조건절
setSortKeys: 소트 키를 지정한다.

JdbcPagingItemReader 작성하기

    @Bean
    public JdbcPagingItemReader<Customer> jdbcPagingItemReader() throws Exception {

        Map<String, Object> parameterValue = new HashMap<>();
        parameterValue.put("age", 20);

        return new JdbcPagingItemReaderBuilder<Customer>()
                .name("jdbcPagingItemReader")
                .fetchSize(CHUNK_SIZE)
                .dataSource(dataSource)
                .rowMapper(new BeanPropertyRowMapper<>(Customer.class))
                .queryProvider(queryProvider())
                .parameterValues(parameterValue)
                .build();
    }

전체 코드

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

import com.schooldevops.springbatch.batchsample.jobs.models.Customer;
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.database.JdbcPagingItemReader;
import org.springframework.batch.item.database.Order;
import org.springframework.batch.item.database.PagingQueryProvider;
import org.springframework.batch.item.database.builder.JdbcPagingItemReaderBuilder;
import org.springframework.batch.item.database.support.SqlPagingQueryProviderFactoryBean;
import org.springframework.batch.item.file.FlatFileItemReader;
import org.springframework.batch.item.file.FlatFileItemWriter;
import org.springframework.batch.item.file.builder.FlatFileItemReaderBuilder;
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.ClassPathResource;
import org.springframework.core.io.FileSystemResource;
import org.springframework.jdbc.core.BeanPropertyRowMapper;
import org.springframework.transaction.PlatformTransactionManager;

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

@Slf4j
@Configuration
public class JdbcPagingReaderJobConfig {

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

    @Autowired
    DataSource dataSource;

    @Bean
    public JdbcPagingItemReader<Customer> jdbcPagingItemReader() throws Exception {

        Map<String, Object> parameterValue = new HashMap<>();
        parameterValue.put("age", 20);

        return new JdbcPagingItemReaderBuilder<Customer>()
                .name("jdbcPagingItemReader")
                .fetchSize(CHUNK_SIZE)
                .dataSource(dataSource)
                .rowMapper(new BeanPropertyRowMapper<>(Customer.class))
                .queryProvider(queryProvider())
                .parameterValues(parameterValue)
                .build();
    }

    @Bean
    public PagingQueryProvider queryProvider() throws Exception {
        SqlPagingQueryProviderFactoryBean queryProvider = new SqlPagingQueryProviderFactoryBean();
        queryProvider.setDataSource(dataSource);  // DB 에 맞는 PagingQueryProvider 를 선택하기 위함
        queryProvider.setSelectClause("id, name, age, gender");
        queryProvider.setFromClause("from customer");
        queryProvider.setWhereClause("where age >= :age");

        Map<String, Order> sortKeys = new HashMap<>(1);
        sortKeys.put("id", Order.DESCENDING);

        queryProvider.setSortKeys(sortKeys);

        return queryProvider.getObject();
    }

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


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

        return new StepBuilder("customerJdbcPagingStep", jobRepository)
                .<Customer, Customer>chunk(CHUNK_SIZE, transactionManager)
                .reader(jdbcPagingItemReader())
                .writer(customerFlatFileItemWriter())
                .build();
    }

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

결과

output 디렉토리에 customer_new_v1.csv 파일이 생성되고 내용은 다음과 같다.

KIDO9	41	M
KIDO8	33	F
KIDO7	25	M
KIDO4	40	M
KIDO3	30	M
KIDO2	20	M

JdbcPagingItemReader WrapUp

지금까지 JdbcPagingItemReader를 이용하여 데이터베이스의 내용을 읽어서, 파일로 저장해 보았다.
JdbcPagingItemReader은 데이터소스를 주입받고, QueryProvider를 이용하여 쿼리를 작성하는 코드도 작성해보았다.
페이징 처리의 성능을 개선하면, 매우 큰 데이터도 효율적으로 수행할 수 있다.
페이징은 pageSize로 지정된 수만큼 읽어오고, 이를 청크로 전달된다는 것도 확인할 수 있었다.

JdbcBatchItemWriter 개요

JdbcBatchItemWriter Spring Batch에서 제공하는 ItemWriter 인터페이스를 구현하는 클래스이다.
데이터를 JDBC를 통해 데이터베이스에 저장하는 데 사용된다.

JdbcBatchItemWriter 구성 요소

DataSource: 데이터베이스 연결 정보를 지정한다.
SqlStatementCreator: INSERT 쿼리를 생성하는 역할을 한다.
PreparedStatementSetter: INSERT 쿼리의 파라미터 값을 설정하는 역할을 한다.
ItemSqlParameterSourceProvider: Item 객체를 기반으로 PreparedStatementSetter에 전달할 파라미터 값을 생성하는 역할을 한다.

장점

데이터베이스 연동: JDBC를 통해 다양한 데이터베이스에 데이터를 저장할 수 있다.
성능: 대량의 데이터를 빠르게 저장할 수 있다.
유연성: 다양한 설정을 통해 원하는 방식으로 데이터를 저장할 수 있다.

단점

설정 복잡성: JDBC 설정 및 쿼리 작성이 복잡할 수 있다.
오류 가능성: 설정 오류 시 데이터 손상 가능성이 있다.

JdbcBatchItemWriter 샘플 코드

application.yaml 파일 작성하기

spring:
  datasource:
    url: jdbc:mysql://localhost:3306/testdb?useUnicode=true&characterEncoding=utf8&clusterInstanceHostPattern=?&zeroDateTimeBehavior=CONVERT_TO_NULL&allowMultiQueries=true
    username: root
    password: root1234
  batch:
    job:
      name: JDBC_BATCH_WRITER_CHUNK_JOB

테이블 생성하기

create table testdb.customer2
(
    id     int auto_increment primary key,
    name   varchar(100) null,
    age    int          null,
    gender varchar(10)  null
);

JdbcBatchItemWriter 작성하기

  @Bean
    public JdbcBatchItemWriter<Customer> flatFileItemWriter() {

        return new JdbcBatchItemWriterBuilder<Customer>()
                .dataSource(dataSource)
                .sql("INSERT INTO customer2 (name, age, gender) VALUES (?, ?, ?)")
                .itemSqlParameterSourceProvider(new CustomerItemSqlParameterSourceProvider())
                .build();
    }

SqlPatameterSourceProvider 작성하기

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

import com.schooldevops.springbatch.batchsample.jobs.models.Customer;
import org.springframework.batch.item.database.ItemSqlParameterSourceProvider;
import org.springframework.jdbc.core.namedparam.BeanPropertySqlParameterSource;
import org.springframework.jdbc.core.namedparam.SqlParameterSource;

public class CustomerItemSqlParameterSourceProvider implements ItemSqlParameterSourceProvider<Customer> {
    @Override
    public SqlParameterSource createSqlParameterSource(Customer item) {
        return new BeanPropertySqlParameterSource(item);
    }
}

전체 코드

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


import com.schooldevops.springbatch.batchsample.jobs.models.Customer;
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.database.JdbcBatchItemWriter;
import org.springframework.batch.item.database.builder.JdbcBatchItemWriterBuilder;
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 JdbcBatchItemJobConfig {

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

    @Autowired
    DataSource dataSource;

    @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 JdbcBatchItemWriter<Customer> flatFileItemWriter() {

        return new JdbcBatchItemWriterBuilder<Customer>()
                .dataSource(dataSource)
                .sql("INSERT INTO customer2 (name, age, gender) VALUES (:name, :age, :gender)")
                .itemSqlParameterSourceProvider(new CustomerItemSqlParameterSourceProvider())
                .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(flatFileItemWriter())
                .build();
    }

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

결과

작업을 완료하면 customer2 테이블에 다음과 같이 데이터가 인서트 된 것을 확인할 수 있다.

id,name,age,gender
1,unclebae,40,Male
2,superman,45,Male
3,WonderWoman,30,Female

JdbcPagingItemReader WrapUp

JdbcBatchItemWriter 를 작성하였다.
jdbc를 이용하고, sqlquery를 이용하여 직접 쿼리를 작성하고, 파라미터로 Customer 객체의 필드 이름을 세팅한결과 정상적으로 데이터가 저장되었다.
간단한 처리, 혹은 쿼리를 직접 작성해야하는 경우 유용한 JdbcBatchItemWriter에 대해 알수 있었다.

#Spring #Programming Language

등록

DEVOTEE를 활성화 시키면
지금 작성한 댓글에 AI가 댓글을 달아줍니다.

익명

KIDO 님의 최신 블로그

PIP를 대체하는 UV 사용법 가이드

이제 PIP 대신에 uv를 사용하기 UV는 Rust로 작성된 매우 빠른 Python 패키지 및 프로젝트 관리자입니다. 이 튜토리얼에서는 UV의 설치부터 기본적인 사용법까지 단계별로 알아보겠습니다. UV 소개 UV는 Python 패키지 관리와 프로젝트 관리를 위한 현대적인 도구입니다. 주요 특징은 다음과 같습니다: ⚡️ pip보다 10-100배 빠른 속도 🐍 Python 버전 관리 지원 🗂️ 종합적인 프로젝트 관리 기능 💾 디스크 공간 효율적인 전역 캐시 🖥️ macOS, Linux, Windows 지원 관련 github: https://github.com/astral-sh/uv PIP vs UV 비교 작업 pip + venv uv 프로젝트 초기화 mkdir...
[SpringBatch 연재 11] Job 및 Step 마다 시작/종료시 특정 작업 수행하기

Job 및 Step 마다 시작/종료시 특정 작업 수행하기 스프링 배치 프레임워크는 Job내부에 여러개의 Step을 수행할 수 있는 구조라는 것을 지금까지 아티클을 통해서 익숙해 졌을 것이다. 그렇다면 Job 시작하고, 종료할 때 특정 작업(보통 처리된 통계 내역등을 확인하여 리포트를 보내야하는 경우)이 있는 경우라면 어떻게 할 수 있을까? 혹은 Step 을 시작하고, 종료할 때 특정 로깅을 해야한다면 어떻게 할 수 있을까? 이번 아티클에서 스프링 배치는 Listener 를 이용할 수 있다. Job Listener 이용하기 JobListener은 Job이 시작/종료 할때마다 특정 이벤트를 수행하고 싶은경우 사용된다. 아래와 같이 Listener를 생성해볼 것이다. ...
[SpringBatch 연재 10] 스프링배치 플로우 컨트롤 하기

SpringBatch Flow Controller 개요 Spring Batch에서 배치 수행 Flow 컨트롤은 여러 Step을 정의하고 조건에 따라 순서대로 실행하거나 특정 Step을 건너뛸 수 있도록 하는 기능이다. 이는 FlowBuilder API를 사용하여 설정된다. Flow 컨트롤 방법 next: 현재 Step이 성공적으로 종료되면 다음 Step으로 이동한다. from: 특정 Step에서 현재 Step으로 이동한다. on: 특정 ExitStatus에 따라 다음 Step을 결정한다. to: 특정 Step으로 이동한다. stop: 현재 Flow를 종료한다. end: FlowBuilder를 종료한다. Flow 컨트롤 샘플 코드 next next는 Start 스텝...
[SpringBatch 연재 09] 입맛에 맞는 배치 처리를 위한 Custom ItemReader/ItemWriter 구현방법 알아보기

이번시간에는 Custom ItemReader와 ItemWriter에 대해서 알아볼 것이다. 스프링 배치는 사전에 구현된 내장 ItemReader와 ItemWriter를 가지고 있어서 일반적인 케이스는 모두 수용이 가능하다. 그러나 특이 케이스나, 비즈니스 로직에 딱 맞는 배치를 수행하기 위해서 Customize가 필요하다. - 우선 QueryDsl을 이용하여 Paging하여 데이터베이스의 값을 읽어 들이는 QuerydslPagingItemReader를 구현해 볼 것이다. - 그리고 CustomItemWriter를 구현하여 타 서비스를 호출하는 샘플을 만들어 보자. QuerydslPagingItemReader 개요 Querydsl은 SpringBatch의 공식 It...

목록으로

DEVOCEAN

로그인 선택

미리보기

커뮤니티

글 등록

임시저장함

데보션 블로그 게재 요청

[SpringBatch 연재 05] JdbcPagingItemReader로 DB내용을 읽고, JdbcBatchItemWriter로 DB에 쓰기

JdbcPagingItemReader 개요

JdbcPagingItemReader 주요 구성 요소

JdbcPagingItemReader 샘플코드

Customer 클래스 생성

쿼리 Provider 생성하기

JdbcPagingItemReader 작성하기

전체 코드

결과

JdbcPagingItemReader WrapUp

JdbcBatchItemWriter 개요

JdbcBatchItemWriter 구성 요소

장점

단점

JdbcBatchItemWriter 샘플 코드

application.yaml 파일 작성하기

테이블 생성하기

JdbcBatchItemWriter 작성하기

SqlPatameterSourceProvider 작성하기

전체 코드

결과

JdbcPagingItemReader WrapUp

KIDO 님의 최신 블로그