스프링 배치 청크 프로세스 활용 - ItemReader

FlatFileItemReader - 개념 및 API 소개

기본개념

• 2차원 데이터로 표현된 유형의 파일을 처리하는 ItemReader
• 일반적으로 고정 위치로 정의된 데이터 필드나 특수 문자에 의해 구별된 데이터의 행을 읽는다.
• Resource 와 LineMapper 2가지 요소가 필요하다.

Resource = 스프링에서 제공하는 인터페이스로 FileSystemResource , ClassPathResource 구현체가 있다.

LineMapper

• 파일의 라인 한줄을 Object로 변환해서 FlatFileItemReader 로 리턴한다.
• 단순히 문자열을 받기 때문에 문자열을 토큰화 해서 객체로 매핑하는 과정이 필요하다.
• LineTokenizer와 FieldSetMapper 를 사용해서 처리한다.

FieldSet

• 라인을 필드로 구분해서 만든 배열 토큰을 전달하면 토큰 필드를 참조할 수 있도록 한다.
• JDBC의 ResultSet과 유사하다.

LineTokenizer

• 입력받은 라인을 FieldSet으로 변환해서 리턴한다.
• 파일마다 형식이 다르기 때문에 문자열을 FieldSet으로 변환하는 작업을 추상화시켜야 한다.

FieldSetMapper

• FieldSet 객체를 받아서 원하는 객체로 매핑해서 리턴한다.
• JdbcTemplate 의 RowMapper 와 동일한 패턴을 사용한다.

 Customer 클래스에 존재하는 필드에 맵핑을 시켜주는곳이 FiledSetMapper이다.
읽어온 자료를 바탕으로 맵핑이 이뤄진다.

 

FixedLength로 파일을 구분자를 기준으로 앞쪽 몇 개만 읽어올 수 있다.

strict을 통해 라인 parsing 실패해도 걍 넘어가게 할 수 있음

 

---

FlatFileItemReader - delimetedlinetokenizer

한개의 라인의 String을 구분자 기준으로 나누어 토큰화 하는 방식

DelimitedLineTokenizer라는 기본구현체는 comma를 기준으로 나눈다.

 

---

FlatFileItemReader - fixedlengthtokenizer

기본개념

• 한개의 라인의 String을 사용자가 설정한 고정길이 기준으로 나누어 토큰화 하는 방식
• 범위는 문자열 형식으로 설정할 수 있다.
• "1-4" , "1-3,4-6,7" = 이처럼 지정할 수 있고 7 이후 범위가 지정되어 있지 않으면 열 끝까지 읽어버린다.

Range에 설정한 만큼 1-5 , 6-10 , 11-14 하나의 String을 쪼개서 token으로 만든후에 FieldSet으로 mapper에게 전달된다.

 

---

FlatFileItemReader - Exception Handling

기본 개념

• 라인을 읽거나 토큰화 할때 발생하는 Parsing 예외를 처리할 수 있도록 예외 계층 제공
• 토큰화 검증을 엄격하게 적용하지 않도록 설정하면 Parsing 예외가 발생하지 않도록 할 수 있다.

LineTokenizer의 Strict을 False로 하면 Tokenizer가 라인 길이를 검증하지 않고 예외를 던지지않는다.

FieldSet은 성공적으로 return 되면서 2번째 범위 값은 빈 토큰을 가지게 된다.

 

---

XML StaxEventItemReader - 개념 및 API 소개

• DOM 방식
  문서 전체를 메모리에 로드한 후 Tree 형태로 만들어서 데이터를 처리하는 방식, pull 방식
  엘리멘트 제어는 유연하나 문서크기가 클 경우 메모리 사용이 많고 속도가 느림
• SAX (simple api xml) 방식
  문서의 항목을 읽을 때 마다 이벤트가 발생하여 데이터를 처리하는 push 방식
  메모리 비용이 적고 속도가 빠른 장점은 있으나 엘리멘트 제어가 어려움
• StAX 방식 (Streaming API for XML)
  DOM 과 SAX 의 장점과 단점을 보완한 API 모델로서 push 와 pull 을 동시에 제공함
  XML 문서를 읽고 쓸 수 있는 양방향 파서기 지원
  XML 파일의 항목에서 항목으로 직접 이동하면서 Stax 파서기를 통해 구문 분석
  
  유형
• Iterator API 방식
  XMLEventReader 의 nextEvent() 를 호출해서 이벤트 객체를 가지고 옴
  이벤트 객체는 XML 태그 유형 (요소, 텍스트, 주석 등)에 대한 정보를 제공함
• Cursor API 방식
  JDBC Resultset 처럼 작동하는 API 로서 XMLStreamReader 는 XML 문서의 다음 요소로 커서를 이동한다
  커서에서 직접 메서드를 호출하여 현재 이벤트에 대한 자세한 정보를 얻는다

 

• Spring-OXM 
  스프링의 Object XML Mapping 기술로 XML 바인딩 기술을 추상화함  (Object , xml 으로 와리가리 하게 해줌)
  
  
• Marshaller
  marshall – 객체를 XML 로 직렬화하는 행위
• Unmashhaller 
  unmarshall – XML 을 객체로 역직렬화하는 행위
• Marshaller 와 Unmarshaller 바인딩 기능을 제공하는 오픈소스로 JaxB2, Castor, XmlBeans, Xstream 등이 있다
  
  
• 스프링 배치는 특정한 XML 바인딩 기술을 강요하지 않고 Spring OXM 에 위임한다
  바인딩 기술을 제공하는 구현체를 선택해서 처리하도록 한다.
  
  
  
• Spring Batch XML 
  
  스프링 배치에서는 StAX 방식으로 XML 문서를 처리하는 StaxEventItemReader 를 제공한다
  XML 을 읽어 자바 객체로 매핑하고 자바 객체를 XML 로 쓸 수 있는 트랜잭션 구조를 지원

Fragment를 읽을때는 pull 방식 (미리 읽을 수있게 Fragment node로 만듬) 
객체로 바인딩 할때는 push 방식  한번에 한개씩 push

Dom Tree를 Spring OXM이 사용한 XStreamMarshaller를 통해서 원하는 객체로 mapping (Trade) 하게 된다.

---

JsonItemReader

@Bean
public JsonItemReader<Customer> customItemReader(){
    return new JsonItemReaderBuilder<Customer>()
            .jsonObjectReader(new JacksonJsonObjectReader<>(Customer.class))
            .resource(new ClassPathResource("customer.json"))
            .name("jsonItemReader")
            .build();
}

---

DB - Cursor & Paging 이해

기본개념

• 배치 어플리케이션은 실시간적 처리가 어려운 대용량 데이터를 다루며 이 때 DB I/O 의 성능문제와 메모리 자원의 효율성 문제를 해결할 수 있어야 한다.
• 스프링 배치에서는 대용량 데이터 처리를 위한 두 가지 해결방안을 제시하고 있다

Cursor Based 처리

• JDBC ResultSet 의 기본 메커니즘을 사용
• 현재 행에 커서를 유지하며 다음 데이터를 호출하면 다음 행으로 커서를 이동하며 데이터 반환이 이루어지는 Streaming 방식의 I/O 이다
• ResultSet이 open 될 때마다 next() 메소드가 호출 되어 Database의 데이터가 반환되고 객체와 매핑이 이루어진다.
• DB Connection 이 연결되면 배치 처리가 완료될 때 까지 데이터를 읽어오기 때문에 DB와 SocketTimeout을 충분히 큰 값으로 설정 필요
• 모든 결과를 메모리에 할당하기 때문에 메모리 사용량이 많아지는 단점이 있다
• Connection 연결 유지 시간과 메모리 공간이 충분하다면 대량의 데이터 처리에 적합할 수 있다 (fetchSize 조절)

Paging Based 처리

• 페이징 단위로 데이터를 조회하는 방식으로 Page Size 만큼 한번에 메모리로 가지고 온 다음 한 개씩 읽는다.
• 한 페이지를 읽을때마다 Connection을 맺고 끊기 때문에 대량의 데이터를 처리하더라도 SocketTimeout 예외가 거의 일어나지 않는다
• 시작 행 번호를 지정하고 페이지에 반환시키고자 하는 행의 수를 지정한 후 사용 – Offset, Limit
• 페이징 단위의 결과만 메모리에 할당하기 때문에 메모리 사용량이 적어지는 장점이 있다
• Connection 연결 유지 시간이 길지 않고 메모리 공간을 효율적으로 사용해야 하는 데이터 처리에 적합할 수 있다

 

---

DB - JdbcCursorItemReader

• Cursor 기반의 JDBC 구현체로서 ResultSet과 함께 사용되며 Datasource에서 Connection을 얻어와서 SQL을 실행
• thread safe하지 않음 => read가 중복해서 같은 row를 가지고 오게 될 수도 있음

JdbcCursorItemReader 는 ItemStream을 구현하고 있어서 read하는 순간 부모클래스의 open을 알아서 호출하면서 초기화를 진행하게 된다.

@Bean
public JdbcCursorItemReader<Customer> customItemReader() {
    return new JdbcCursorItemReaderBuilder<Customer>()
            .name("jdbcCursorItemReader")
            .fetchSize(10)
            .sql("select * from customer where firstName like ? order by lastName")
            .beanRowMapper(Customer.class)
            .queryArguments("A%")
            .maxItemCount(3)
            .currentItemCount(2)
            .maxRows(100)
            .dataSource(dataSource)
            .build();
}

? 에 queryArguments 가 들어가게된다.

---

DB - JpaCursorItemReader

Spring Batch 4.3이후 부터 Cursor기반의 JPA 구현체로서 EntityManagerFactory 객체가 필요하며 쿼리는 JPQL을 사용함

Iterator에서 next를 호출하는 시점에는 이미 open()메소드에서 필요한 db 작업을 다 해놓아서 결과물들이 존재하는 상태이다.

JDBCCursorItemReader는 Resultset next() 를 호출할떄마다 db connection을 타고 가져오는것이었다.

---

DB - JdbcPagingItemReader

기본개념

• Paging 기반의 JDBC 구현체로서 쿼리에 시작 행 번호 (offset) 와 페이지에서 반환 할 행 수 (limit)를 지정해서 SQL 을 실행한다
• 스프링 배치에서 offset과 limit을 PageSize에 맞게 자동으로 생성해 주며 페이징 단위로 데이터를 조회할 때 마다 새로운 쿼리가 실행한다
• 페이지마다 새로운 쿼리를 실행하기 때문에 페이징 시 결과 데이터의 순서가 보장될 수 있도록 order by 구문이 작성되도록 한다
• 멀티 스레드 환경에서 Thread 안정성을 보장하기 때문에 별도의 동기화를 할 필요가 없다

PagingQueryProvider

• 쿼리 실행에 필요한 쿼리문을 ItemReader 에게 제공하는 클래스
• 데이터베이스마다 페이징 전략이 다르기 때문에 각 데이터 베이스 유형마다 다른 PagingQueryProvider 를 사용한다
• Select 절, from 절, sortKey 는 필수로 설정해야 하며 where, group by 절은 필수가 아니다

:firstname 에 Alika가 들어가게 된다. cursor를 쓰는 방법과 다르게 내부적으로 암묵락을 잡고 있어 동기화 걱정 할 필요 없다.

---

DB - JpaPagingItemReader

Paging 기반의 JPA 구현체로서 EntityManagerFactory 객체가 필요하며 쿼리는 JPQL을 사용한다.

 

@Bean
public JpaPagingItemReader<Customer> customItemReader() {

    return new JpaPagingItemReaderBuilder<Customer>()
            .name("jpaPagingItemReader")
            .entityManagerFactory(entityManagerFactory)
            .pageSize(10)
            .queryString("select c from Customer c join fetch c.address")
            .build();
}

---

ItemReaderAdapter

배치 Job안에서 이미 있는 DAO 나 다른 서비스를 ItemReader 안에서 사용하고자 할 때 위임하는 역할을 한다.

기존의 다른 서비스로 부터 ItemRead를 수행하고 싶다. 기존 다른 서비스가 데이터를 제공하는것임 , reflection으로 호출해서 ItemReader처럼 동작하게 해준다는 의미이다.