Spring Batch ItemReader, Item Writer

개요

Delegate Pattern Registering

1. 직접 래핑해서 등록하기

@Bean
public SynchronizedItemStreamReader<Foo> reader() {
    SynchronizedItemStreamReader<Foo> reader = new SynchronizedItemStreamReader<>();
    reader.setDelegate(new FooReader());
    return reader;
}

2. 빈으로 등록된 delegate를 주입받아 등록하기

@Bean
public FooReader fooReader() {
    return new FooReader();
}

@Bean
public SynchronizedItemStreamReader<Foo> reader(FooReader delegate) {
    SynchronizedItemStreamReader<Foo> reader = new SynchronizedItemStreamReader<>();
    reader.setDelegate(delegate);
    return reader;
}

Flat Files

FieldSet

String[] tokens = new String[]{"foo", "1", "true"};
FieldSet fs = new DefaultFieldSet(tokens);
String name = fs.readString(0);
int value = fs.readInt(1);
boolean booleanValue = fs.readBoolean(2);

FlatFileItemReader

@Bean
public FlatFileItemReader<Customer> reader() {
    FlatFileItemReader<Customer> reader = new FlatFileItemReader<>();
    reader.setResource(new ClassPathResource("customer.csv"));
    reader.setLineMapper(new DefaultLineMapper<>() {{
        setLineTokenizer(new DelimitedLineTokenizer() {{
            setNames("firstName", "lastName", "birthDate");
        }});
        setFieldSetMapper(new BeanWrapperFieldSetMapper<>() {{
            setTargetType(Customer.class);
        }});
    }});
    return reader;
}

FlatFileItemReader의 주요 구성 요소

1. Resource

reader.setResource(new FileSystemResource("input/customers.csv"));

1. LineMapper<T>

DefaultLineMapper<Customer> lineMapper = new DefaultLineMapper<>();

1. LineTokenizer

DelimitedLineTokenizer tokenizer = new DelimitedLineTokenizer();
tokenizer.setNames("id", "name", "email");

1. FieldSetMapper<T>

Simple Delimited File Reading Example

ID,lastName,firstName,position,birthYear,debutYear
"AbduKa00,Abdul-Jabbar,Karim,rb,1974,1996",
"AbduRa00,Abdullah,Rabih,rb,1975,1999",
"AberWa00,Abercrombie,Walter,rb,1959,1982",
"AbraDa00,Abramowicz,Danny,wr,1945,1967",
"AdamBo00,Adams,Bob,te,1946,1969",
"AdamCh00,Adams,Charlie,wr,1979,2003"

1. 먼저 Player 도메인을 정의합니다.

public class Player implements Serializable {

    private String ID;
    private String lastName;
    private String firstName;
    private String position;
    private int birthYear;
    private int debutYear;

    public String toString() {
        return "PLAYER:ID=" + ID + ",Last Name=" + lastName +
            ",First Name=" + firstName + ",Position=" + position +
            ",Birth Year=" + birthYear + ",DebutYear=" +
            debutYear;
    }

    // setters and getters...
}

1. LineMapper의 LineTokenizer를 정의하여 File의 각 줄을 자릅니다.
앞서 설명한것처럼 LineTokenizer는 크게 두가지 성격으로 나눌 수 있습니다.
1. CSV와 같은 파일의 경우 DelimitedLineTokenizer
2. 고정 길이 파일의 경우 FixedLengthTokenizer

USER;Smith;Peter;;T;20014539;F
LINEA;1044391041ABC037.49G201XX1383.12H
LINEB;2134776319DEF422.99M005LI

@Bean
public PatternMatchingCompositeLineMapper orderFileLineMapper() {
	PatternMatchingCompositeLineMapper lineMapper =
		new PatternMatchingCompositeLineMapper();

	Map<String, LineTokenizer> tokenizers = new HashMap<>(3);
	tokenizers.put("USER*", userTokenizer());
	tokenizers.put("LINEA*", lineATokenizer());
	tokenizers.put("LINEB*", lineBTokenizer());

	lineMapper.setTokenizers(tokenizers);

	Map<String, FieldSetMapper> mappers = new HashMap<>(2);
	mappers.put("USER*", userFieldSetMapper());
	mappers.put("LINE*", lineFieldSetMapper());

	lineMapper.setFieldSetMappers(mappers);

	return lineMapper;
}

1. FieldSetMapper를 사용하여 Player 도메인에 바인딩합니다.

protected static class PlayerFieldSetMapper implements FieldSetMapper<Player> {
    public Player mapFieldSet(FieldSet fieldSet) {
        Player player = new Player();

        player.setID(fieldSet.readString(0));
        player.setLastName(fieldSet.readString(1));
        player.setFirstName(fieldSet.readString(2));
        player.setPosition(fieldSet.readString(3));
        player.setBirthYear(fieldSet.readInt(4));
        player.setDebutYear(fieldSet.readInt(5));

        return player;
    }
}

FlatFileItemReader<Player> itemReader = new FlatFileItemReader<>();
itemReader.setResource(new FileSystemResource("resources/players.csv"));
DefaultLineMapper<Player> lineMapper = new DefaultLineMapper<>();
//DelimitedLineTokenizer defaults to comma as its delimiter
lineMapper.setLineTokenizer(new DelimitedLineTokenizer());
lineMapper.setFieldSetMapper(new PlayerFieldSetMapper());
itemReader.setLineMapper(lineMapper);
itemReader.open(new ExecutionContext());
Player player = itemReader.read();

@Bean
public FieldSetMapper fieldSetMapper() {
	BeanWrapperFieldSetMapper fieldSetMapper = new BeanWrapperFieldSetMapper();

	fieldSetMapper.setPrototypeBeanName("player");

	return fieldSetMapper;
}

@Bean
@Scope("prototype")
public Player player() {
	return new Player();
}

tokenizer.setColumns(new Range[] { new Range(1, 5),
                                   new Range(6, 10),
                                   new Range(11, 15) });
try {
    tokenizer.tokenize("12345");
    fail("Expected IncorrectLineLengthException");
}
catch (IncorrectLineLengthException ex) {
    assertEquals(15, ex.getExpectedLength());
    assertEquals(5, ex.getActualLength());
}

FlatFileItemWriter

FlatFileItemWriter의 주요 구성 요소

1. LineAggregator<T>
LineAggregator는 객체를 문자열로 변환하는 역할을 합니다. 즉, 각 아이템을 파일에 기록하기 전에 문자열 형태로 변환하는 로직을 정의합니다.
• aggregate(T item): 주어진 아이템을 문자열로 변환합니다.
예를 들어, CSV 파일에 데이터를 기록하려면 각 필드를 쉼표로 구분하여 하나의 문자열로 만들어야 합니다. 이러한 작업을 LineAggregator에서 수행합니다.

2. FieldExtractor<T>
FieldExtractor는 객체의 필드 값을 추출하는 역할을 합니다. LineAggregator와 함께 사용되어 객체를 문자열로 변환하는 데 필요한 필드 값을 제공합니다.
• extract(T item): 주어진 아이템에서 필드 값을 추출하여 배열로 반환합니다.
예를 들어, BeanWrapperFieldExtractor는 Java Bean의 프로퍼티 값을 추출하는 데 사용됩니다.

3. Resource
출력 대상 파일을 지정합니다. 예를 들어, FileSystemResource를 사용하여 로컬 파일 시스템의 경로를 지정할 수 있습니다.

Simplified File Writing Example

@Bean
public FlatFileItemWriter itemWriter() {
	return  new FlatFileItemWriterBuilder<Foo>()
           			.name("itemWriter")
           			.resource(new FileSystemResource("target/test-outputs/output.txt"))
           			.lineAggregator(new PassThroughLineAggregator<>())
           			.build();
}

public class PassThroughLineAggregator<T> implements LineAggregator<T> {

    public String aggregate(T item) {
        return item.toString();
    }
}

FieldExtractor

1. Item을 Writer에게 제공

2. Item의 Fields를 배열로 변환

3. 결과 배열을 한 줄로 집계

@Bean
public FlatFileItemWriter<CustomerCredit> itemWriter(Resource outputResource) {
    return new FlatFileItemWriterBuilder<CustomerCredit>()
            .name("customerCreditWriter")
            .resource(outputResource)
            .delimited()
            .delimiter("|")
            .names("name", "credit") // Getter 기준
            .build();
}

@Bean
public FlatFileItemWriter<CustomerCredit> itemWriter(Resource outputResource) {
    BeanWrapperFieldExtractor<CustomerCredit> fieldExtractor = new BeanWrapperFieldExtractor<>();
    fieldExtractor.setNames(new String[] {"name", "credit"});
    fieldExtractor.afterPropertiesSet();

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

    return new FlatFileItemWriterBuilder<CustomerCredit>()
            .name("customerCreditWriter")
            .resource(outputResource)
            .lineAggregator(lineAggregator)
            .build();
}

@Bean
public FlatFileItemWriter<CustomerCredit> itemWriter(Resource outputResource) throws Exception {
	return new FlatFileItemWriterBuilder<CustomerCredit>()
				.name("customerCreditWriter")
				.resource(outputResource)
				.formatted()
				.format("%-9s%-2.0f")
				.names(new String[] {"name", "credit"})
				.build();
}

1. 직접 구성 방식 (BeanWrapperFieldExtractor 명시)

@Bean
public FlatFileItemWriter<CustomerCredit> itemWriter(Resource outputResource) throws Exception {
	BeanWrapperFieldExtractor<CustomerCredit> fieldExtractor = new BeanWrapperFieldExtractor<>();
	fieldExtractor.setNames(new String[] {"name", "credit"});
	fieldExtractor.afterPropertiesSet();

	FormatterLineAggregator<CustomerCredit> lineAggregator = new FormatterLineAggregator<>();
	lineAggregator.setFormat("%-9s%-2.0f");
	lineAggregator.setFieldExtractor(fieldExtractor);

	return new FlatFileItemWriterBuilder<CustomerCredit>()
				.name("customerCreditWriter")
				.resource(outputResource)
				.lineAggregator(lineAggregator)
				.build();
}

JSON Files

JsonItemReader

주요 구성 요소

• Resource: 읽을 JSON 파일을 나타내는 Spring의 리소스 객체입니다.

• JsonObjectReader<T>: JSON 객체를 Java 객체로 변환하는 역할을 합니다. 현재 Jackson과 Gson 기반의 두 가지 구현체가 제공됩니다:
• JacksonJsonObjectReader
• GsonJsonObjectReader

JsonItemReader Example

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

JsonFileItemWriter

주요 구성 요소

• WritableResource: 출력할 JSON 파일을 나타내는 Spring의 리소스 객체입니다.

• JsonObjectMarshaller<T>: Java 객체를 JSON 문자열로 변환하는 역할을 합니다. 현재 Jackson과 Gson 기반의 두 가지 구현체가 제공됩니다:
• JacksonJsonObjectMarshaller
• GsonJsonObjectMarshaller

JsonFileItemWriter Example

@Bean
public JsonFileItemWriter<Trade> jsonItemWriter() {
    return new JsonFileItemWriterBuilder<Trade>()
        .jsonObjectMarshaller(new JacksonJsonObjectMarshaller<>())
        .resource(new FileSystemResource("output-trades.json"))
        .name("tradeJsonItemWriter")
        .build();
}

주의사항

1. Partitioning 사용
• 각 파티션이 서로 다른 JSON 파일을 읽고 쓰도록 설정하면, 스레드 간 충돌이 발생하지 않습니다.

2. Step 단위로 인스턴스를 분리
• 각 Step에서 JsonItemReader를 새로 생성하여 사용하면 스레드 간 공유 상태를 피할 수 있습니다.
• 이 때에도 Scope("step") 또는 @StepScope 어노테이션을 사용할 수 있습니다.

Database

• 커서 기반

• 페이징 기반

ItemReader

Cursor-based ItemReader

• JdbcCursorItemReader: JDBC를 사용하여 데이터를 스트리밍 방식으로 읽습니다. ResultSet을 통해 데이터를 한 행씩 처리하며, 메모리에 전체 데이터를 로드하지 않기 때문에 메모리 효율성이 높습니다.

• HibernateCursorItemReader: Hibernate를 사용하여 데이터를 커서 방식으로 읽습니다. Hibernate의 ScrollableResults를 활용하여 데이터를 순차적으로 처리합니다.

• StoredProcedureItemReader: 저장 프로시저를 호출하여 결과를 커서 방식으로 읽습니다. 데이터베이스에 정의된 저장 프로시저를 통해 데이터를 가져옵니다.

• 커서 기반 방식은 데이터베이스 커넥션을 Step의 시작부터 끝까지 유지하므로, 장시간 커넥션 유지로 인한 리소스 소모나 잠금 문제에 주의해야 합니다.

Paging-based ItemReader

• JdbcPagingItemReader: JDBC를 사용하여 데이터를 페이지 단위로 읽습니다. 쿼리에서 OFFSET과 LIMIT을 사용하여 데이터를 분할하여 가져옵니다.

• HibernatePagingItemReader: JPA를 사용하여 데이터를 페이지 단위로 읽습니다. JPA의 페이징 기능을 활용하여 데이터를 분할하여 가져옵니다.

ItemWriter

• JdbcBatchItemWriter: JDBC를 사용하여 데이터를 일괄적으로 삽입합니다. PreparedStatement를 사용하여 배치로 데이터를 처리하므로 성능이 우수합니다.

• HibernateItemWriter: Hibernate를 사용하여 데이터를 저장합니다. Hibernate의 Session을 통해 엔티티를 저장하거나 갱신합니다.

마무리

Reference

Database :: Spring Batch

Like most enterprise application styles, a database is the central storage mechanism for batch. However, batch differs from other application styles due to the sheer size of the datasets with which the system must work. If a SQL statement returns 1 million rows, the result set probably holds all returned results in memory until all rows have been read. Spring Batch provides two types of solutions for this problem:

https://docs.spring.io/spring-batch/reference/readers-and-writers/database.html