[Spring batch] Step 2 - CSV 파일을 읽어서 출력하기

먼저 CSV파일을 준비한다.

https://data.fivethirtyeight.com/ 사이트에서 "Club Soccer Predictions" 항목을 다운받아서 사용해 보았다.

CSV 파일 읽기 => 도메인객체(VO)로 매핑 => 출력하기
위의 작업을 수행하기 위해서는 세부적으로 아래의 작업을 수행해야 한다.

* 배치프로그램 실행시 파라미터를 사용하여 파일정보를 읽어들인다.
* CSV에는 파일의 메타정보가 존재하지 않는다. 메타정보를 정의하는 Domain 객체(VO)를 생성한다.
* ItemReader 인터페이스를 사용하여 FlatFile을 읽는다. 읽을때에는 delimiter 를 쉼표(,)로 지정한다.
* ItemWriter 인터페이스를 사용하여 읽어들인 정보를 콘솔 출력한다.
* Step를 만들어서 ItemReader -> ItemWriter 를 수행한다.
* Job을 만들어서 Step를 수행한다.

일단 flatfile(csv)의 형태는 다음과 같다. 
spi_global_rankings.csv

rank,prev_rank,name,league,off,def,spi
1,1,Manchester City,Barclays Premier League,2.9,0.2,93.73
2,3,Liverpool,Barclays Premier League,2.92,0.25,93.15
3,2,Bayern Munich,German Bundesliga,3.62,0.56,93.01
4,4,Chelsea,Barclays Premier League,2.39,0.29,88.69
5,6,Real Madrid,Spanish Primera Division,2.58,0.5,86.63

위의 형태를 읽어오기 위해서 아래와 같이 도메인 객체를 만들어 보았다.
Ranking.java

@Data
public class Ranking {
    //rank,prev_rank,name,league,off,def,spi
    private int rank;
    private int prev_rank;
    private String name;
    private String league;
    private float off;
    private float def;
    private float spi;
}

이제는 파일을 읽어올 차례이다.
파일을 읽어오려면 FlatFile의 필드와 도메인객체(VO)간의 매핑을 해 주어야 한다.
아래와 같이 매핑코드를 작성하였다.
RankingFieldSetMapper.java

public class RankingFieldSetMapper implements FieldSetMapper<Ranking> {

    @Override
    public Ranking mapFieldSet(FieldSet fieldSet) throws BindException {
        Ranking ranking = new Ranking();
        //rank,prev_rank,name,league,off,def,spi
        ranking.setRank(fieldSet.readInt("rank"));
        ranking.setPrev_rank(fieldSet.readInt("prev_rank"));
        ranking.setName(fieldSet.readString("name"));
        ranking.setLeague(fieldSet.readString("league"));
        ranking.setOff(fieldSet.readFloat("off"));
        ranking.setDef(fieldSet.readFloat("def"));
        ranking.setSpi(fieldSet.readFloat("spi"));

        return ranking;
    }
}

그리고 나서 FlatFile의 delimiter(구분자)를 정해 주어야 하는데, csv파일을 보통 쉼표로 정의한다. 이것은 Tokenizer 에서 담당한다. 그래서 아래와 같이 Tokenizer 를 구현해 주었다.

    @Bean
    public PatternMatchingCompositeLineMapper lineTokenizer() {

        // lineTokenizers 에 파싱정보 전달
        Map<String, LineTokenizer> lineTokenizers = new HashMap<>(1);
        lineTokenizers.put("*",new RankingLineTokenizer());

        //필드셋 매퍼에 도메인객체 매핑정보를 전달
        Map<String, FieldSetMapper> fieldSetMappers = new HashMap<>(1);
        fieldSetMappers.put("*",new RankingFieldSetMapper());

        //패턴매칭라인매퍼에 토크나이저와 필드셋매퍼 정보를 전달
        PatternMatchingCompositeLineMapper lineMappers = new PatternMatchingCompositeLineMapper();
        lineMappers.setTokenizers(lineTokenizers);
        lineMappers.setFieldSetMappers(fieldSetMappers);

        return lineMappers;
    }

순서를 보자면 먼저 토크나이저를 만든다(멀티토크나이징일 경우 Map을 사용해야 한다. 위의 경우는 하나라서 필요없지만 책의 예제가 이렇게 설명되어서 그냥 씀)
그리고 필드 매퍼를 만든다.
패턴매칭컴포지트라인매퍼에 담은값을 셋팅하여 넘겨준다. 헉헉...(csv파일의 유형이 멀티일 경우 Map에 담아서 넘겨준다)

흐흥, 파싱 정보를 설정해주어야 겠지요?? 파싱정보를 RankingLineTokenizer.java에 구현해두었다. (책에는 DelimitedLineTokenizer 를 사용하는 것으로 소개되었지만 클래스로 분리를 위해 직접 구현하는 방식으로 변경)

    public class RankingLineTokenizer implements LineTokenizer {
    private String delimiter = ",";

    private String[] names = new String[]{
            "rank","prev_rank","name","league","off","def","spi"
    };
    private FieldSetFactory fieldSetFactory = new DefaultFieldSetFactory();

    @Override
    public FieldSet tokenize(String record) {
        String[] fields = record.split(delimiter);
        List<String> parsedFields = new ArrayList<>();
        for(int i=0; i<fields.length;i++){
            parsedFields.add(fields[i]);
        }
        return fieldSetFactory.create(parsedFields.toArray(new String[0]),names);
    }
}

이제 남은 일은 ItemReader (아이템 읽기), ItemWriter(아이템 쓰기) 그리고 수행할 스텝과 잡을 만드는 일이다.

    //파일을 읽기위한 ItemReader를 선언한다.
    @Bean
    @StepScope
    public FlatFileItemReader<Ranking> rankingItemReader(@Value("#{jobParameters['filename']}") Resource inputFile) {
        return new FlatFileItemReaderBuilder<Ranking>()
                .name("rankingItemReader")
                .lineMapper(lineTokenizer())
                .linesToSkip(1) //csv파일을 첫번째줄은 건너뛴다 (default값은 0)
                .resource(inputFile)
                .build();
    }

    @Bean
    public ItemWriter<Ranking> itemWriter(){
        return (items) -> items.forEach(System.out::println);
    }
    
    //스텝을 만든다.
    @Bean
    public Step step(){
        return this.stepBuilderFactory.get("step1")
                .<Ranking, Ranking>chunk(10)
                .reader(rankingItemReader(null))
                .writer(itemWriter())
                .build();
    }

    //잡을 만든다.
    @Bean
    public Job job() {
        return this.jobBuilderFactory.get("job1")
                .start(step())
                .build();
    }

    public static void main(String[] args) {
        SpringApplication.run(BatchApplication.class, args);
    }

spi_global_rankings.csv 파일은 첫번째 라인이 필드명을 선언하고 있어서 lineToSkip 에 1의 값을 주었다.
외부 파라미터를 받아서 수행하기 위해 @Value("#{jobParameters['filename']}") 을 주었다.
출력은 도메인객체에 대한 콘솔출력이다 (다음 포스트에서 DBMS로 저장하도록 ItemWriter개선 예정)

Run/Debug Confiurations 의 Program arguments 에
"filename=file:d:\spi_global_rankings.csv foo=1"
위와 같이 넣고 실행하면 아래와 같이 출력되는 것을 볼 수 있다. (foo=1 은 스프링 배치가 동일 파라미터로 실행시 실행안되기 때문에 임의로 넣었다)

  .   ____          _            __ _ _
 /\\ / ___'_ __ _ _(_)_ __  __ _ \ \ \ \
( ( )\___ | '_ | '_| | '_ \/ _` | \ \ \ \
 \\/  ___)| |_)| | | | | || (_| |  ) ) ) )
  '  |____| .__|_| |_|_| |_\__, | / / / /
 =========|_|==============|___/=/_/_/_/
 :: Spring Boot ::                (v2.6.4)

2022-03-21 00:57:33.180  INFO 11564 --- [           main] com.sunnier.batch.BatchApplication       : Starting BatchApplication using Java 1.8.0_202 on DESKTOP-AHN11RT with PID 11564 (C:\git_repo\batch\target\classes started by sunni in C:\git_repo\batch)
2022-03-21 00:57:33.192  INFO 11564 --- [           main] com.sunnier.batch.BatchApplication       : No active profile set, falling back to 1 default profile: "default"
2022-03-21 00:57:34.098  INFO 11564 --- [           main] com.zaxxer.hikari.HikariDataSource       : HikariPool-1 - Starting...
2022-03-21 00:57:34.166  INFO 11564 --- [           main] com.zaxxer.hikari.HikariDataSource       : HikariPool-1 - Start completed.
2022-03-21 00:57:34.360  INFO 11564 --- [           main] o.s.b.c.r.s.JobRepositoryFactoryBean     : No database type set, using meta data indicating: MYSQL
2022-03-21 00:57:34.485  INFO 11564 --- [           main] o.s.b.c.l.support.SimpleJobLauncher      : No TaskExecutor has been set, defaulting to synchronous executor.
2022-03-21 00:57:34.595  INFO 11564 --- [           main] com.sunnier.batch.BatchApplication       : Started BatchApplication in 1.927 seconds (JVM running for 2.723)
2022-03-21 00:57:34.595  INFO 11564 --- [           main] o.s.b.a.b.JobLauncherApplicationRunner   : Running default command line with: [filename=file:d:\spi_global_rankings.csv, foo=1]
2022-03-21 00:57:34.696  INFO 11564 --- [           main] o.s.b.c.l.support.SimpleJobLauncher      : Job: [SimpleJob: [name=job1]] launched with the following parameters: [{filename=file:d:\spi_global_rankings.csv, foo=1}]
2022-03-21 00:57:34.755  INFO 11564 --- [           main] o.s.batch.core.job.SimpleStepHandler     : Executing step: [step1]
Ranking(rank=1, prev_rank=1, name=Manchester City, league=Barclays Premier League, off=2.9, def=0.2, spi=93.73)
Ranking(rank=2, prev_rank=3, name=Liverpool, league=Barclays Premier League, off=2.92, def=0.25, spi=93.15)
Ranking(rank=3, prev_rank=2, name=Bayern Munich, league=German Bundesliga, off=3.62, def=0.56, spi=93.01)
Ranking(rank=4, prev_rank=4, name=Chelsea, league=Barclays Premier League, off=2.39, def=0.29, spi=88.69)
Ranking(rank=5, prev_rank=6, name=Real Madrid, league=Spanish Primera Division, off=2.58, def=0.5, spi=86.63)
Ranking(rank=6, prev_rank=5, name=Ajax, league=Dutch Eredivisie, off=3.02, def=0.74, spi=86.49)
Ranking(rank=7, prev_rank=9, name=Barcelona, league=Spanish Primera Division, off=2.48, def=0.54, spi=84.77)
Ranking(rank=8, prev_rank=7, name=Paris Saint-Germain, league=French Ligue 1, off=2.6, def=0.67, spi=83.52)
Ranking(rank=9, prev_rank=10, name=RB Leipzig, league=German Bundesliga, off=2.63, def=0.72, spi=82.91)

.
.
.
.
.
.
.
.
.
Ranking(rank=639, prev_rank=636, name=Oldham Athletic, league=English League Two, off=0.24, def=2.55, spi=5.59)
Ranking(rank=640, prev_rank=640, name=Scunthorpe, league=English League Two, off=0.2, def=2.91, spi=3.73)
2022-03-21 00:57:35.505  INFO 11564 --- [           main] o.s.batch.core.step.AbstractStep         : Step: [step1] executed in 745ms
2022-03-21 00:57:35.530  INFO 11564 --- [           main] o.s.b.c.l.support.SimpleJobLauncher      : Job: [SimpleJob: [name=job1]] completed with the following parameters: [{filename=file:d:\spi_global_rankings.csv, foo=1}] and the following status: [COMPLETED] in 814ms
2022-03-21 00:57:35.539  INFO 11564 --- [ionShutdownHook] com.zaxxer.hikari.HikariDataSource       : HikariPool-1 - Shutdown initiated...
2022-03-21 00:57:35.553  INFO 11564 --- [ionShutdownHook] com.zaxxer.hikari.HikariDataSource       : HikariPool-1 - Shutdown completed.
Disconnected from the target VM, address: '127.0.0.1:51176', transport: 'socket'

Process finished with exit code 0

배치실행이 잘 되었다... ㅎㅎ

 

 

 추가)
DelimiterLineTokenizer를 사용할 경우 아래와 같이 사용하면 된다.

    // 파싱정보를 설정한다. (RankingLineTokenizer.java에 구현함)
    @Bean
    public DelimitedLineTokenizer rankingLineTokenizer() {
        DelimitedLineTokenizer lineTokenizer = new DelimitedLineTokenizer();
        lineTokenizer.setNames("rank","prev_rank","name","league","off","def","spi");
        return lineTokenizer;
    }

물론 토크나이저를 불러오려면 아래와 같이 사용한다.

        // lineTokenizers 에 파싱정보 전달
        Map<String, LineTokenizer> lineTokenizers = new HashMap<>(1);
        lineTokenizers.put("*",rankingLineTokenizer());