[대용량 이관] bookRatingStep 개선

 

 

[bookRatingStep]을 만들었다. 300만 데이터 넣는데에 211초가 걸렸고, 문득 성능 테스트하는데 노트북 사양을 full 로 쓰면 조금 아이러니한 느낌이 들어서 메모리를 제한해보고 테스트해보기로 했다.

 

 

---

 

수정사항

1. 메모리 설정

-Xms512m -Xmx1g -XX:+PrintGCDetails -XX:+PrintGCDateStamps -Xloggc:gc.log

옵션	설명
-Xms512m	JVM의 초기 힙 메모리 크기를 512MB로 설정
-Xmx1g	JVM의 최대 힙 메모리 크기를 1GB로 제한
-XX:+PrintGCDetails	GC가 발생할 때 어떤 세대에서 얼마나 수집되었는지 상세하게 로그 출력
-XX:+PrintGCDateStamps	GC 로그에 날짜와 시간 정보 포함
-Xloggc:gc.log	GC 로그를 gc.log 파일로 저장

 

힙 메모리
자바에서 new로 만든 객체들(예: DTO, List, Map 등) 은 모두 힙 메모리에 올라간다.
JVM은 힙에 있는 객체를 관리하며, 일정 주기마다 GC(Garbage Collector) 가 이를 정리함.

 

processor 에서 객체로 변환해서 전달하는 단순한 로직이지만, chunk 사이즈에 따라 n개의 객체가 메모리에 유지된 상태에서 한 번에 처리하기 때문에 반복되면 JVM 힙에서 많은 객체가 생성, 소멸, 참조되면서 GC 부담이 생길 수 있다고 판단했다.

 

 

2. insert 속도 측정

@Bean
    public ItemWriter<BookRatingDto> bookRatingWriter() {
        String tableName = "book_rating";
        JdbcBatchItemWriter<BookRatingDto> delegate = new JdbcBatchItemWriterBuilder<BookRatingDto>()
                .dataSource(dataSource)
                .sql(StepConstant.getSqlQuery(tableName, bookRatingColumns, false))
                .beanMapped()
                .build();
        delegate.afterPropertiesSet();
        return new ItemWriter<>() {
            @Override
            public void write(Chunk<? extends BookRatingDto> chunk) throws Exception {
                long start = System.currentTimeMillis();
                delegate.write(chunk);
                long end = System.currentTimeMillis();
                log.info("📦 INSERTED {} rows in {} ms", chunk.getItems().size(), end - start);
            }
        };
    }

Insert 속도를 확인하기위해 ItemWriter 를 오버라이드했다. 단순히 start, end 계산이라서 어렵진 않다.

 

java.lang.NullPointerException: Cannot invoke "org.springframework.batch.item.database.ItemPreparedStatementSetter.setValues(Object, java.sql.PreparedStatement)" because "this.itemPreparedStatementSetter" is null at org.springframework.batch.item.database.JdbcBatchItemWriter.lambda$write$0(JdbcBatchItemWriter.java:189) ~[spring-batch-infrastructure-5.2.2.jar:5.2.2] at org.springframework.jdbc.core.JdbcTemplate.execute(JdbcTemplate.java:658) ~[spring-jdbc-6.2.6.jar:6.2.6] at org.springframework.jdbc.core.JdbcTemplate.execute(JdbcTemplate.java:701) ~[spring-jdbc-6.2.6.jar:6.2.6] at org.springframework.batch.item.database.JdbcBatchItemWriter.write(JdbcBatchItemWriter.java:187) ~[spring-batch-infrastructure-5.2.2.jar:5.2.2] at io.github.haeun.batch.job.task.book.step.BookRatingStep$1.write(BookRatingStep.java:114) ~[main/:na] at org.springframework.batch.core.step.item.SimpleChunkProcessor.writeItems(SimpleChunkProcessor.java:203) ~[spring-batch-core-5.2.2.jar:5.2.2] at org.springframework.batch.core.step.item.SimpleChunkProcessor.doWrite(SimpleChunkProcessor.java:170) ~[spring-batch-core-5.2.2.jar:5.2.2] at org.springframework.batch.core.step.item.SimpleChunkProcessor.write(SimpleChunkProcessor.java:297) ~[spring-batch-core-5.2.2.jar:5.2.2]
Truncating long message before update of StepExecution, original message is: java.lang.NullPointerException: Cannot invoke "org.springframework.batch.item.database.ItemPreparedStatementSetter.setValues(Object, java.sql.PreparedStatement)" because "this.itemPreparedStatementSetter" is null at org.springframework.batch.item.database.JdbcBatchItemWriter.lambda$write$0(JdbcBatchItemWriter.java:189)

"delegate.afterPropertiesSet();" 라는 부분을 넣지 않으면 위 에러를 만나게 된다. 

 

JdbcBatchItemWriter의 afterPropertiesSet()은 다음을 체크하고, 설정 누락이 있으면 예외를 발생시켜 조기에 알려준다.

• dataSource가 설정 되어있는지
• sql이 null이 아닌지
• itemSqlParameterSourceProvider가 있는지
• beanMapped() 또는 columnMapped() 설정이 되어있는지

JdbcBatchItemWriter를 Bean으로 등록할 경우에는 자동으로 호출하나, 수정한 로직처럼 ItemWriter로 리턴하게되면 호출되지 않기때문에 수동으로 호출시켜줘야한다.

 

 

 

 

Chunk 

1. chunk 1000

insert 속도는 20~40ms 정도였다. 오히려 시간체크까지 추가돼서 전보다 느릴 줄 알았는데 첫 테스트보다 시간이 줄었다;

insert 하는데에 시간이 오래 걸리지 않으니 chunk size 를 늘리면서 개선이 가능한지 테스트해보려고 한다.

 

 

2. chunk 3000

insert 속도가 chunk 1000보다 3배 이상 늘어났다...! 여기서 더 늘리면 insert 속도가 더 느릴 것 같지만 궁금해서 테스트해봤다.

 

 

3. chunk 5000

여태까지 했던 것 중에 가장 빠르다. 우연히 빨랐던건지 의아해서 두 번 돌렸는데 오히려 더 빨라졌다.

chunk 1000의 평균 inset속도를 35로 잡으면 조금 더 느린 정도인데 빨라진 이유는 DB에 요청보내는 횟수가  3,000 -> 600 으로 줄어들면서 나온 속도 개선으로 추정된다.

 

 

4. chunk 10000

서버에서 job 하나만 돌리는게 아닌 이상은 10,000 단위까지는 안할 것 같은데, 속도 변화가 궁금해서 해봤다.

큰 차이가 생기지 않았다...

 

 

5. chunk 500

500으로 줄여본 결과 100이랑 2초정도밖에 차이가 나지 않았다. 

 

 

결과

chunk size	Insert	Total(s)
1,000	20~40	189
3,000	100~140	201
5,000	180~190	178
10,000	340~380	176
500	10~15	191

 

chunk size 를 늘림으로써 속도가 줄었으나 그만큼 많은 데이터를 갖고 있다가 Insert 하는 것이기 때문에 부담이 생긴다. 그 부담만큼 속도가 드라마틱하게 변하지 않아서 500~1,000 정도의 사이즈로도 충분해보인다.

견딜 수 있을 정도의 chunk size라면 오히려 늘릴수록 빠를 것으로 예상했었다. 실제로도 줄긴 했지만, 늘린 크기에 비해 시간은 줄지 않아서 늘리는 것만이 답이 아닌 걸 알았다.

 

 

 

Thread

@Bean(name = "stepBookRating")
public Step bookRatingStep() {
    return new StepBuilder("stepBookRating", jobRepository)
            .<Map<String, Object>, BookRatingDto>chunk(500, transactionManager)
            .reader(bookRatingReader())
            .processor(bookRatingProcessor())
            .writer(bookRatingWriter())
            .taskExecutor(taskExecutor())
            .throttleLimit(poolSize)
            .build();
}
@Bean
    public SynchronizedItemStreamReader<Map<String, Object>> bookRatingReader() {
        FlatFileItemReader<Map<String, Object>> delegate = CsvReaderFactory.createReader(
                new FileSystemResource("./Amazon Books Reviews/Books_rating.csv"),
                "bookReader",
                new String[]{"Id", "Title", "Price", "User_id", "profileName", "review/helpfulness", "review/score", "review/time", "review/summary", "review/text"}
        );
        SynchronizedItemStreamReader<Map<String, Object>> reader = new SynchronizedItemStreamReader<>();
        reader.setDelegate(delegate);
        return reader;
    }
@Bean
public TaskExecutor taskExecutor() {
    ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
    executor.setCorePoolSize(poolSize);
    executor.setMaxPoolSize(poolSize);
    executor.setThreadNamePrefix("batch-thread-");
    executor.initialize();
    return executor;
}

spring:
  datasource:
    url: jdbc:mariadb://localhost:3306/batch
    username: user
    password: user
    driver-class-name: org.mariadb.jdbc.Driver
    hikari:
      maximum-pool-size: 10

속도 개선이 가장 확실한 쓰레드를 늘려보기로했다.

taskExecutor를 추가해서 풀사이즈를 지정하고, FlatFileItemReader는 스레드 세이프하지 않기 때문에 동일한 로우를 읽을 위험이 있어서 SynchronizedItemStreamReader 를 사용하도록 했다.

 

 

1. thread 2

chunk 1000으로 테스트 시 40초정도 줄어들었다.

좀 더 늘려보자.

 

 

2. thread 4

 

 

3. thread 10

4와 10의 속도가 똑같다. 이상해서 한번더 돌렸는데 오히려 더 느린 결과가 나왔다.

 

실행시키고 processlist 를 확인해보니 커넥션은 생성이 되었는데, 동시에 Insert 하는 커넥션이 적다. read 가 sysnc로 싱글 스레드로 처리하기 때문에 발생하는 문제이지 않을까 의심이 갔다. 파일이 여러개라면 MultiResourcePartitioner 로 테스트해볼텐데 일단 이건 나중에 해봐야겠다.

 

 

결과

Thread	Chunk	Total(s)
2	1,000	145
4	1,000	92
10	1,000	92

 

병령처리를 하니 chunk 테스트했을 때보다 절반정도로 줄어들었다. 300만건 넣는데 92초면 초당 32,600 넣은 것이기 때문에 빠르다고 볼 수 있다. 하지만 병목현상이 나타나는 것 같은데 이걸 처리해서 좀 더 개선을 하고 싶다는 생각이 든다.