[Spring Batch] ItemReader - CSV, JDBC, JPA

Chunk 기반의 Step에서 ItemReader, ItemProcessor, ItemWriter가 존재한다.

이번 글을 통해 CSV, JDBC, JPA의 데이터를 읽어오는 ItemReader를 소개하고자 한다.

 

CSV File ItemReader

A flat file is any type of file that contains at most two-dimensional (tabular) data. Reading flat files in the Spring Batch framework is facilitated by the class called FlatFileItemReader, which provides basic functionality for reading and parsing flat files. - Spring Batch Docs

 

요약하면 2차원 데이터를 읽으려면 FlatFileItemReader 클래스를 사용한다.

FlatFileItemReader는 "Resource"와 "LineMapper"가 필요하다.

 

• Resource : 말 그대로 자원을 의미한다. 우리는 CSV 파일을 읽을거기 때문에 CSV파일이 Resource라 할 수 있다.
• LineMapper : flat file에서 각 행을 읽어 객체로 매핑해주는 인터페이스다.

 

예시

Test.csv

id,이름,나이,거주지
1,홍길동,24,서울
2,짱구,5,떡잎마을
3,상디,20,올블루

 

위의 csv파일을 FlatFileItemReader를 통해 읽어보겠습니다.

@Entity
@Getter
@NoArgsConstructor
public class Person {

  @Id
  @GeneratedValue(strategy = GenerationType.IDENTITY)
  private int id;
  private String name;
  private String age;
  private String address;

  public Person(String name, String age, String address) {
    this(0, name, age, address);
  }

  public Person(int id, String name, String age, String address) {
    this.id = id;
    this.name = name;
    this.age = age;
    this.address = address;
  }
}
---
@Bean
public Step csvFileStep() throws Exception {
  return stepBuilderFactory.get("csvFileStep")
      .<Person, Person>chunk(10)
      .reader(csvFileItemReader())
      .writer(itemWriter()) // writer는 reader를 통해 읽어온 이름을 출력하게 했습니다.
      .build();
}

 

1. LineMapper에 Tokenizer와 FieldSet을 설정합니다.

private FlatFileItemReader<Person> csvFileItemReader() throws Exception {
  DefaultLineMapper<Person> lineMapper = new DefaultLineMapper<>();
  DelimitedLineTokenizer tokenizer = new DelimitedLineTokenizer();
  tokenizer.setNames("id", "name", "age", "address");
  lineMapper.setLineTokenizer(tokenizer);

  lineMapper.setFieldSetMapper(fieldSet -> {
    int id = fieldSet.readInt("id");
    String name = fieldSet.readString("name");
    String age = fieldSet.readString("age");
    String address = fieldSet.readString("address");

    return new Person(id, name, age, address);
  });

 

LineMapper는 Tokenizer와 Fieldset을 가집니다.

• Tokenizer : 각 줄의 데이터를 구분자를 통해 추출한다. (보통 csv파일은 쉼표로 구분되어 있으므로 쉼표를 구분기호로 사용한다.) 
                      Tokenizer를 통해 추출한 값을 FieldSet으로 변환한다.
• FieldSet :  FieldSet is Spring Batch’s abstraction for enabling the binding of fields from a file resource. It allows developers to work with file input in much the same way as they would work with database input. - docs
  즉, 파일을 통해 필드를 바인딩할 수 있게 하는 추상화이다. (글만 읽어서는 무슨 의미인지 잘 와닿지 않는다.)

어쨌든, LineMapper은 Tokenizer와 FieldSet이 필요하다. 그래서 Tokenizer의 names을 csv 파일의 각 column 명으로 설정한다. 

그리고 fieldSet에서 각 column명으로 데이터를 가져와서 Person 객체를 반환하게 한다.

 

2. FlatFileItemReader를 생성한다. 이때 Builder를 사용해서 생성한다.

FlatFileItemReader<Person> itemReader = new FlatFileItemReaderBuilder<Person>()
    .name("csvFileItemReader")
    .encoding("UTF-8")
    .resource(new ClassPathResource("test.csv"))
    .linesToSkip(1) // csv의 첫 번째 로우인 필드명을 스킵하겠다는 의미
    .lineMapper(lineMapper)
    .build();
itemReader.afterPropertiesSet(); // itemReader에서 필수 설정 값이 정상적으로 설정됐는지 확인해주는 메서드

return itemReader;

 

위에서 작성한 것처럼 FlatFileItemReader는 resource와 lineMapper가 필요하다. 

/resources/test.csv

lineMappper은 위에서 만들어준 DefaultLineMapper를 설정해주고 Resource는 test.csv를 설정해줘야한다. 

test.csv 파일은 resources폴더 하위에 존재한다. ClassPathResource 클래스는 resources 하위에 파일을 읽어올 수 있게 하는 클래스이므로 이 클래스를 사용해서 파일을 읽어온다.

 

이 과정을 통해서 CSV 파일을 FlatFileItemReader를 통해 읽어올 수 있다. 읽어온 값은 다음과 같다.

1,홍길동,24,서울
2,짱구,5,떡잎마을
3,상디,20,올블루

---

JDBC ItemReader

우선 JDBC ItemReader를 알아보기 전에 Cursor 기반 조회와 Paging 기반 조회를 알아봅시다.

Spring의 JdbcTemplate는 분할 처리를 지원하지 않아서 Cursor와 Paging으로 데이터를 읽어옵니다.

• Cursor 기반 조회 (DB Connection 연결 후 Cursor를 1칸씩 옮기면서 데이터를 완료될 때까지 가져온다.)
  • 배치 처리가 완료될 때까지 DB Connection이 연결
  • DB Connection 빈도가 낮아 성능이 좋지만 긴 Connection 유지 시간이 필요하다.
  • 하나의 Connection에서 처리되기 때문에, Thread Safe 하지 않다.
  • 모든 결과를 메모리에 할당하기 때문에, 더 많은 메모리를 사용한다.
• Paging 기반 조회 (개발자가 지정한 PageSize만큼 데이터를 가져온다.)
  • 페이징 단위로 DB Connection을 연결
  • DB Connection 빈도가 높아 비교적 성능은 낮지만 짧은 Connection 유지 시간이 필요하다.
  • 매번 Connection을 하기 때문에 Thread Safe하다.
  • 페이징 단위의 결과만 메모리에 할당하기 때문에, 비교적 더 적은 메모리를 사용한다.

Cursor 기반 조회와 Paging 기반 조회의 구현 코드는 유사하기 때문에 Cursor 기반 조회를 구현해보겠습니다.

 

예시

create table person (
  id bigint primary key auto_increment,
  name varchar(255),
  age varchar(255),
  address varchar(255)
);

insert into person(name, age, address) values('홍길동', '24', '서울');
insert into person(name, age, address) values('짱구', '5', '떡잎마을');
insert into person(name, age, address) values('상디', '20', '올블루');

@Bean
public Step jdbcStep() throws Exception {
  return stepBuilderFactory.get("jdbcStep")
      .<Person, Person>chunk(10)
      .reader(jdbcCursorItemReader())
      .writer(itemWriter())
      .build();
}

 

위의 SQL은 빌드시 실행됩니다.

 

private JdbcCursorItemReader<Person> jdbcCursorItemReader() throws Exception {
  JdbcCursorItemReader<Person> itemReader = new JdbcCursorItemReaderBuilder<Person>()
      .name("jdbcCursorItemReader")
      .dataSource(dataSource)
      .sql("select id, name, age, address from person")
      .rowMapper(((rs, rowNum) -> new Person(rs.getInt(1), rs.getString(2), rs.getString(3),
          rs.getString(4))))
      .build();
  itemReader.afterPropertiesSet();
  return itemReader;
}

---
private final DataSource dataSource; // DataSource DI

 

Cursor 기반 조회에서는 JdbcCursorItemReader 클래스를 사용합니다. JdbcCursorItemReader 객체를 JdbcCursorItemReaderBuilder를 사용하여 생성합니다.

 

코드를 보면 dataSource를 지정하고, sql 쿼리를 설정하고 rowMapper를 설정합니다.

• dataSoource는 application.yml에서 지정한 dataSource를 DI를 통해 가져옵니다.
• sql은 저장된 테이블에서 조회하는 쿼리를 작성합니다
• rowMapper : sql에서 가져온 값을 ResultSet에 담아와서 객체에 저장합니다. 위 코드에선 Person 객체로 저장합니다.
                         rs(resultSet), rowNum(행 번호) => rs.getInt(1) : Column의 1번째인 id의 행을 가져온다는 의미

이후 FlatFileItemReader와 같이 afterPropertiesSet()을 통해 빠진게 없는지 확인합니다.

 

결과

---

JPA ItemReader

JPA ItemReader는 JDBC와 같이 JpaCursorItemReader, JpaPagingItemReader가 존재한다.

JDBC와 거진 비슷하지만 차이점은 dataSource대신 entityManagerFactory를 사용한다는 점, sql대신 queryString을 사용하는데 JDBC는 Native Query지만 JPA는 JPQL을 사용한다는 점이다.

 

예시

@Bean
public Step jpaStep() throws Exception {
  return stepBuilderFactory.get("jpaStep")
      .<Person, Person>chunk(10)
      .reader(jpaCursorItemReader())
      .writer(itemWriter())
      .build();
}

 

private JpaCursorItemReader<Person> jpaCursorItemReader() throws Exception {
  JpaCursorItemReader<Person> itemReader = new JpaCursorItemReaderBuilder<Person>()
      .name("jpaCursorItemReader")
      .entityManagerFactory(entityManagerFactory)
      .queryString("select p from Person p")
      .build();
  itemReader.afterPropertiesSet();
  return itemReader;
}

---
private final EntityManagerFactory entityManagerFactory; // EntityManagerFactory DI

 

마무리

이번에는 Chunk 기반 Step에서 사용되는 ItemReader를 3가지 타입을 통해 알아보았습니다. JDBC와 JPA ItemReader는 Cursor 기반으로만 해봤는데 다음에는 Paging 기반으로 ItemReader를 구현하는 법을 알아보겠습니다. 또한 ItemWriter에 대해서도 3가지 타입으로 알아보겠습니다 !