Spring Batch - ItemProcessor, ItemWriter

FlatFileItemWriter

- 2차원으로 표현된 유형된 파일을 처리하는 ItemWriter

 

LineAggregator

 - Item을 받아서 String으로 변환하여 리턴

- FiledExtractor를 사용해서 처리함

- 구현체

=> PassThroughLineAggregator / DelimitedLineAggregator / FormatterLineAggregator 

FieldExtractor

- 전달받은 Item 객체의 필드를 배열로 만들고, 배열을 합쳐서 문자열을 만들도록 구현하도록 제공하는 인터페이스

- 구현체

=> BeanWrapperFieldExtractor

=> PassThroughFieldExtractor

 

JsonFileItemWriter

- 객체를 받아 JSON String으로 변환하는 역할을 함

    @Bean
    public ItemWriter<Person> writer() {
        return new JsonFileItemWriterBuilder<Person>()
                .jsonObjectMarshaller(new JacksonJsonObjectMarshaller<>())
                .resource(new FileSystemResource(Paths.get("output.json").toFile()))
                .name("personJsonFileItemWriter")
                .build();
    }

 

DB- JdbcBatchItemWriter

- datasource를 지정하고, SQL속성에 실행할 쿼리를 설정

- JDBC의 batch기능을 사용하여 bulk insert, update, delete 방식으로 처리

- 단 건 처리가 아닌 일괄처리여서, 성능에 이점을 가짐

 

public JdbcBatchItemWriter itemWriter() {
	return new JdbcBatchItemWriterBuilder<T>()
    	.name(String name)
        .datasource(Datasource) // DB에 접근하기 위해 DataSource 설정 
        .sql(String sql)	// ItemWriter 쿼리 
        .assertUpdates(boolean) // 예외 발생여부 
        .beanMapped()	// Pojo 기반으로 Insert SQL의 values를 맵핑
        .columnMapped()	// key, value 기반으로 InsertSQL의 values를 맵핑
        .build();
}

 

- ColumnMapped로 설정한 경우 Map타입의 아이템 배치가 처리되고, BeanMapped로 설정한 경우 Pojo 타입의 아이템 배치 처리가 된다.

 

ItemWriterAdapter

- 배치 Job안에서 있는 DAO나 다른 서비스를 ItemWriter안에서 아용하고자 할 때 위임역할을 함

 

ItemProcessor

CompositeItemProcessor

- ItemProcessor들을 연결해서 위임하면 각 ItemProcessor를 실행시킴

- 이전 ItemProcessor반환 값은 다음 ItemProcessor값으로 연결됨 

 

public ItemProcessor itemProcessor () {
	return new CompositeItemProcessorBuilder<>()
    	.delegates(ItemProcessor<?, ?>... delegates)
        .build();
}

    @Bean
    public Step step(StepBuilderFactory stepBuilderFactory, CompositeItemProcessor<String, String> processor) {
        return stepBuilderFactory.get("step")
                .<String, String>chunk(10)
                .reader(reader())
                .processor(processor)
                .writer(writer())
                .build();
    }
    
    @Bean
    public CompositeItemProcessor<String, String> processor() {
        List<ItemProcessor<String, String>> processors = Arrays.asList(new UpperCaseItemProcessor(), new ExclamationItemProcessor());
        CompositeItemProcessor<String, String> compositeProcessor = new CompositeItemProcessor<>();
        compositeProcessor.setDelegates(processors);
        return compositeProcessor;
    }
    
   public static class UpperCaseItemProcessor implements ItemProcessor<String, String> {
        @Override
        public String process(String item) {
            return item.toUpperCase();
        }
    }

    public static class ExclamationItemProcessor implements ItemProcessor<String, String> {
        @Override
        public String process(String item) {
            return item + "!";
        }
    }

 

ClassifierCompositeItemProcessor

- Classifier로 라우팅 패턴을 구현해서, ItemProcessor 구현체에서 하나를 호출하는 역할을 함

 

public ItemProcessor itemProcessor() {
	return new ClassifierCompositeItemProcessorBuilder<>() 
    	.classfier(Classifier)
        .build();
}

 

 

public class NumberProcessor implements ItemProcessor<Object, String> {
    @Override
    public String process(Object item) {
        return "Processed number: " + item;
    }
}

public class StringProcessor implements ItemProcessor<Object, String> {
    @Override
    public String process(Object item) {
        return "Processed string: " + item;
    }
}

public class TypeClassifier implements Classifier<Object, ItemProcessor<Object, String>> {

    private final ItemProcessor<Object, String> numberProcessor;
    private final ItemProcessor<Object, String> stringProcessor;

    public TypeClassifier(ItemProcessor<Object, String> numberProcessor, ItemProcessor<Object, String> stringProcessor) {
        this.numberProcessor = numberProcessor;
        this.stringProcessor = stringProcessor;
    }

    @Override
    public ItemProcessor<Object, String> classify(Object item) {
        if (item instanceof Integer) {
            return numberProcessor;
        } else if (item instanceof String) {
            return stringProcessor;
        } else {
            throw new IllegalArgumentException("Unknown type: " + item.getClass());
        }
    }
}

 

 

    @Bean
    public Step step(StepBuilderFactory stepBuilderFactory, ClassifierCompositeItemProcessor<Object, String> processor) {
        return stepBuilderFactory.get("step")
                .<Object, String>chunk(10)
                .reader(reader())
                .processor(processor)
                .writer(writer())
                .build();
    }
    
    @Bean
    public ClassifierCompositeItemProcessor<Object, String> processor() {
        ClassifierCompositeItemProcessor<Object, String> processor = new ClassifierCompositeItemProcessor<>();
        processor.setClassifier(new TypeClassifier(new NumberProcessor(), new StringProcessor()));
        return processor;
    }

- processor() 메서드는 ClassifierCompositeItemProcessor를 정의하고, TypeClassifier를 사용해서 데이터 타입에 따라 적절한 ItemProcessor를 선택함

 

 

출처 - 

인프런 스프링 배치 강의