Spring Batch JdbcBatchItemWriter

ItemReader는 다양한 매체의 아이템들을 읽을 수 있습니다.

ItemWriter는 출력이 가능합니다.

 

이 JdbcBatchItemWriter는 jdbc를 사용해 db에 update, insert, delete 쿼리를 날릴 수 있습니다.

이름에 Batch가 있는 이유는  쿼리를 일괄 실행할 수 있기 때문입니다.

한 건씩 하는 것이아닌  bulk insert/update/delete를 할 수 있다는 것인데요

즉 여러개의 쿼리를 한 번에 직접 데이터베이스에 쿼리 하는 것입니다.

 

먼저 mysql설정을 조금 손봐야합니다.

bulk연산을 사용하기 위해 url설정을 추가해 줍니다.

jdbc-url: jdbc:mysql://127.0.01:3306/spring_batch?characterEncoding=UTF-8&serverTimezone=UTC&rewriteBatchedStatements=true

rewriteBatchedStatements=true // mysql 설정

 

bulk는 jpa를 구현한 하이버네이트의 도움을 받아야 합니다.

ddl-auto는 해당 엔티티가 데이터베이스에 존재하지 않으면 테이블을 생성해주고 해당 엔티티가 변경된 경우는 테이블을 update를 해줍니다. 다만 jpa지원 안 해주는 경우가 있으니 테이블을 따로 만들어서 진행했다.

 

 

@Configuration
@Slf4j
public class ItemWriterConfiguration {

    private final JobBuilderFactory jobBuilderFactory;
    private final StepBuilderFactory stepBuilderFactory;
    private final DataSource dataSource;

    public ItemWriterConfiguration(JobBuilderFactory jobBuilderFactory, StepBuilderFactory stepBuilderFactory, DataSource dataSource) {
        this.jobBuilderFactory = jobBuilderFactory;
        this.stepBuilderFactory = stepBuilderFactory;
        this.dataSource = dataSource;
    }

    @Bean
    public Job itemWriterJob() throws Exception {

        return this.jobBuilderFactory.get("itemWriterJob")
                .incrementer(new RunIdIncrementer())
                .start(this.csvItemWriterStep())
                .next(this.jdbcBatchItemWriterStep())
                .build();

    }

    @Bean
    public Step csvItemWriterStep() throws Exception {

        return this.stepBuilderFactory.get("itemWriterStep")
                .<Person,Person>chunk(10)
                .reader(itemReader())
                .writer(csvFileItemWriter())
                .build();
    }

    @Bean
    public Step jdbcBatchItemWriterStep(){

        return stepBuilderFactory.get("jdbcBatchItemWriterStep")
                .<Person,Person>chunk(10)
                .reader(itemReader())
                .writer(jdbcBatchItemWriter())
                .build();
    }

    private ItemWriter<Person> jdbcBatchItemWriter() {
        JdbcBatchItemWriter<Person> itemWriter = new JdbcBatchItemWriterBuilder<Person>()
                .dataSource(dataSource)
                .itemSqlParameterSourceProvider(new BeanPropertyItemSqlParameterSourceProvider<>())//Person클래스를 자동으로 파라미터로 생성할 수 있는 설정
                .sql("insert into person(name,age,address) values(:name, :age, :address)")
                .build();

        itemWriter.afterPropertiesSet();

        return itemWriter;


    }

    private ItemWriter<Person> csvFileItemWriter() throws Exception {
        BeanWrapperFieldExtractor<Person> fieldExtractor = new BeanWrapperFieldExtractor<>();
        fieldExtractor.setNames(new String[]{"id","name","age","address"}); //필드명

        DelimitedLineAggregator<Person> lineAggregator = new DelimitedLineAggregator<>();
        lineAggregator.setDelimiter(",");
        lineAggregator.setFieldExtractor(fieldExtractor);

        FlatFileItemWriter<Person> itemWriter = new FlatFileItemWriterBuilder<Person>()
                .name("csvFileItemWriter")
                .encoding("UTF-8")
                .resource(new FileSystemResource("output/test-output.csv")) //파일 생성하기 위함
                .lineAggregator(lineAggregator)
                .headerCallback(writer -> writer.write("id,이름,나이,거주지")) //header
                .footerCallback(writer -> writer.write("------------footer-----\n"))
                .append(true)
                .build();

        itemWriter.afterPropertiesSet();

        return itemWriter;
    }

    private ItemReader<Person> itemReader() {
        return new CustomItemReader<>(getItems());
    }

    private List<Person> getItems() {

        List<Person> items = new ArrayList<>();

        for (int i = 0; i < 100; i++) {
            items.add(new Person(i + 1, "test_name", "test age", "test_address"));
        }

        return items;
    }
}

실행 시 다음과 같다.

 

다음과 같이 데이터베이스에 들어간 것을 확인할 수 있다.