Jun, Log

이번 시간에는 스프링 배치 구동 시 파라미터 입력과 파라미터 검증을 위한 validator를 학습 하였습니다.

앞에서의 학습으로 인해서 조금 더 수월한 학습을 진행하였습니다.

job, step 그리고 tasklet을 활용하여 만든 배치를 살펴 보겠습니다.

private final JobBuilderFactory jobBuilderFactory;
private final StepBuilderFactory stepBuilderFactory;

@Bean
public Job advancedJob(Step advancedStep) {
    return jobBuilderFactory.get("advancedJob")
            .incrementer(new RunIdIncrementer())
            //.validator(new LocalDateParameterValidator("targetDate"))
            .start(advancedStep)
            .build();
}

@JobScope
@Bean
public Step advancedStep(Tasklet advancedTasklet) {
    return stepBuilderFactory.get("advancedStep")
            .tasklet(advancedTasklet)
            .build();
}

@StepScope
@Bean
public Tasklet advancedTasklet(
		@Value("#{jobParameters['targetDate']}") String targetDate) {
        
    return (contribution, chunkContext) -> {
        log.info("[AdvancedJobConfig] targetDate = " + targetDate);
        log.info("[AdvancedJobConfig] excuted advancedTasklet");
        
        return RepeatStatus.FINISHED;
    };

}

작업 내용이 없는 job으로써 배치가 수행만 되겠습니다. 그리고 tasklet에 보면 targetDate라는 파라미터를 입력 받는데요. 파라미터 추가를 위해서는 배치 작업 실행 시 다음 예시 처럼 파라미터를 추가를 해주시면 되겠습니다.

--spring.batch.job.names=advancedJob -targetDate=2022-02-07

targetDate라는 파라미터를 추가를 해주었습니다. 실행을 하면 스텝 내에서 파라미터 값을 받아 오신걸 확인 할 수 있습니다.

다음으로 파라미터 검증을 위한 validator을 만들어 보겠습니다.

public class LocalDateParameterValidator implements JobParametersValidator {

    private String parameterName;

    @Override
    public void validate(JobParameters parameters) throws JobParametersInvalidException {
        String localDate = parameters.getString(parameterName);

        if (!StringUtils.hasText(localDate)) {
            throw new JobParametersInvalidException(parameterName + "가 빈 문자열이거나 존재하지 않습니다.");
        }

        try {
            LocalDate.parse(localDate);
        } catch (DateTimeParseException e) {
            throw new JobParametersInvalidException(parameterName + "가 날짜 형식의 문자열이 아닙니다.");
        }
    }
}

validator을 활용하면 step 단계에서 파라미터를 검증 하는 것이 아닌, job 단계에서 배치 수행 전에 검증을 할 수 있습니다. 검증 작업은 빠른면 빠를 수록 좋겠습니다.

JobParametersValidator를 구현 해주시면 되겠습니다. 파라미터를 받고 파라미터의 대한 검증 코드를 작성해주세요.

그리고 Job 메서드에 validator을 등록해주시면 되겠습니다. 첫번째 코드에서는 validator 주석을 풀어주세요 :)

잘못된 파라미터가 들어오면 위와 같은 예외가 발생하겠습니다.

이번 시간에는 스프링 배치에서의 테스트 코드 작성을 학습 하였습니다.

1. 테스트 환경은 h2 데이터베이스를 사용하기 위해서 application.yml 파일 내용을 추가해주세요.

spring:
  config:
    activate:
      on-profile: test
  jpa:
    database: h2

테스트 코드에서 @ActiveProfiles("test") 어노테이션을 통해서 테스트 환경을 구성하겠습니다.

2. 테스트 내에서도 Batch 구동을 위해 Config 파일을 만들었습니다.

@Configuration
@EnableBatchProcessing
@EnableAutoConfiguration
public class BatchTestConfig {
}

test 코드 내에 위의 코드를 작성해주세요. 배치 작업을 위한 설정 어노테이션을 작성하였습니다.

3. 테스트 코드 작성

@SpringBatchTest    // JobLauncherTestUtils 사용을 위해서 필수
@SpringBootTest
@ExtendWith(SpringExtension.class)
@ActiveProfiles("test")
@ContextConfiguration(classes = {HelloJobConfig.class, BatchTestConfig.class})
public class HelloJobConfigTest {

    @Autowired
    private JobLauncherTestUtils jobLauncherTestUtils;

    @Test
    public void success() throws Exception {
        // 실행이 잘되는지 확인
        // when
        JobExecution execution = jobLauncherTestUtils.launchJob();

        // then
        Assertions.assertEquals(execution.getExitStatus(), ExitStatus.COMPLETED);
    }
}

실행이 잘되는지만 확인을 하였습니다.

지금은 HelloJobConfig를 어노테이션에 등록함으로써 HelloJob이 Job으로 등록이 되었습니다.

잡 런쳐는 스프링일 통해서 HelloJob 잡이 등록이 되었는데요. 혹시나 다른 Job도 여기서 한번에 테스트를 하신다고 ContextConfiguration에 여러개의 잡을 등록 하신다면 빈이 여러개 있어 에러가 발생하겠습니다.

step을 구성하는 요소는 크게 2가지가 있습니다. tasklet 방식과, itemXXX방식을 사용하여 처리하는 방식입니다.

앞에 포스트에서는 tasklet을 사용하여 hello world를 봤습니다.

이번 시간에는 ItemReader / ItemProcessor / ItemWriter 을 사용하는 예제를 살펴보겠습니다.

A라는 테이블에서 데이터를 읽어온 후에 B라는 테이블에 데이터를 입력하는 내용입니다.

1. 데이터를 읽은 후에 데이터 저장을 할 수 있도록 테이블을 생성

-- 데이터를 읽을 타겟 테이블
CREATE TABLE `test`.`plain_text` (
  `id` INT NOT NULL AUTO_INCREMENT,
  `text` VARCHAR(100) NOT NULL,
  PRIMARY KEY (`id`));
  
 -- 읽은 데이터를 저장할 타겟 테이블
  CREATE TABLE `test`.`house_text` (
  `id` INT NOT NULL AUTO_INCREMENT,
  `text` VARCHAR(100) NOT NULL,
  PRIMARY KEY (`id`));

-- 기본 데이터 입력
INSERT INTO `test`.`plain_text` (`text`) VALUES ('hi');
INSERT INTO `test`.`plain_text` (`text`) VALUES ('good');
INSERT INTO `test`.`plain_text` (`text`) VALUES ('good!!');
INSERT INTO `test`.`plain_text` (`text`) VALUES ('bye');
INSERT INTO `test`.`plain_text` (`text`) VALUES ('hehe');
INSERT INTO `test`.`plain_text` (`text`) VALUES ('hello!!');

테이블 생성과, 읽어 올 데이터를 생성 하였습니다.

2. 데이터베이스를 읽을 수 있도록 도메인 및 레파지토리 생성

도메인

@Entity
@Getter
@Setter
@DynamicUpdate
@AllArgsConstructor
@NoArgsConstructor
@Table(name="plain_text")
public class PlainText {

    @Id
    @GeneratedValue(strategy = GenerationType.IDENTITY)
    private Integer id;

    @Column(nullable=false)
    private String text;
}

테이블을 생성한 기반으로 도메인 클래스를 만들었습니다.

레파지토리 생성

public interface PlainTextRepository extends JpaRepository<PlainText, Integer> {
    Page<PlainText> findBy(Pageable pageable);
}

데이터를 읽을 수 있도록 레파지토리 생성.

result_text 도메인과 레파지토리도 plain_text와 같이 생성해주세요.

3. Job 생성

@Configuration
@RequiredArgsConstructor
public class PlainTextJobConfig {

    private final JobBuilderFactory jobBuilderFactory;

    private final StepBuilderFactory stepBuilderFactory;

    private final PlainTextRepository plainTextRepository;

    private final ResultTextRepository resultTextRepository;

    @Bean
    public Job plainTextJob(Step plainTextStep) {
        return jobBuilderFactory.get("plainTextJob")
                .incrementer(new RunIdIncrementer())
                .start(plainTextStep)
                .build();
    }

    @JobScope
    @Bean
    public Step plainTextStep(
            ItemReader plainTextReader,
            ItemProcessor plainTextProcessor,
            ItemWriter plainTextWriter) {
        return stepBuilderFactory.get("plainTextStep")
                .<PlainText, String>chunk(5)
                .reader(plainTextReader)
                .processor(plainTextProcessor)
                .writer(plainTextWriter)
                .build();
    }

    @StepScope
    @Bean
    public RepositoryItemReader<PlainText> plainTextReader() {
        return new RepositoryItemReaderBuilder<PlainText>()
                .name("plainTextReader")
                .repository(plainTextRepository)
                .methodName("findBy")
                .pageSize(5)
                .arguments(List.of())   // 전달할 파라미터 있으면 리스트로 넘겨주기.
                .sorts(Collections.singletonMap("id", Sort.Direction.DESC))
                .build();
    }

    @StepScope
    @Bean
    public ItemProcessor<PlainText, String> plainTextProcessor() {
        return new ItemProcessor<PlainText, String>() {
            @Override
            public String process(PlainText item) throws Exception {
                return "processed: " + item.getText();
            }
        };
    }

    @StepScope
    @Bean
    public ItemWriter<String> plainTextWriter() {
        return items -> {
            items.forEach(
            	item -> resultTextRepository.save(new ResultText(null, item))
            );
            System.out.println("===== chunk is finished ====");
        };
    }
}

Job과 Step을 생성하고, Step에는 ItemReader / ItemProcessor / ItemWriter 을 구현하여 연결 하였습니다.

Job이 실행 되고 -> Step이 실행되면서 ItemReader / ItemProcessor / ItemWriter 구성이 각각 수행이 되겠습니다.

배치 실행 결과)

Chunk is finished가 2번이 출력된 모습을 보실 수 있습니다. chunk 사이즈를 5로 하였기 때문에 데이터는 5개씩 처리가 되겠습니다. 총 데이터는 6개가 존재합니다. 5개가 처리되고 나머지 1개도 처리가 되기 때문에 wirter는 총 2번이 호출이 되었습니다.

result_text 실행 결과)

타겟 테이블에 우리가 원하는 데이터가 입력이 된 모습을 보실 수 있습니다.

절대 경로 사용을 위해서 루트 경로에 .env 파일을 만들고 NODE_PATH=src 입력을 하였는데 에러가 발생하였습니다. 더이상 이 구문은 지원을 안한다고 합니다.

절대 경로 설정

1. jsconfig.json 파일을 루트 경로에 만들어 주세요.

2. 아래 코드 입력

{
    "compilerOptions": {
        "baseUrl": "src"
    },
    "include": [
        "src"
    ]
}

3. 이제 컴포넌트에서 상대 경로가 아닌 절대 경로를 사용해주셔도 되겠습니다.

필터링(distinct, filter)

스트림 사용시 필터링 방법을 사용하여 데이터를 걸러낼 수 있습니다.

public static void main(String[] args) {
    List<String> names = Arrays.asList("홍길동", "신용권", "감자바", "신용권", "신민철");

    names.stream()
        .distinct()
        .forEach(n->System.out.println(n));
    System.out.println();

    names.stream()
        .filter(n->n.startsWith("신"))
        .forEach(n->System.out.println(n));
    System.out.println();

    names.stream()
        .distinct()
        .filter(n->n.startsWith("신"))
        .forEach(n->System.out.println(n));
}

첫번째는 distinct()함수를 통해서 중복을 제거 하였습니다.

두번째는 filter() 함수를 통해서 신으로 시작하는 값만 필터링 하였으며, 세번째는 중복을 제거 하고 필터링을 적용 하였습니다.

매핑(map)

매핑은 스트림의 요소들을 다른 요소로 대체를 합니다.

flatMap()

public static void main(String[] args) {
    List<String> inputList1 = Arrays.asList("java8 lambda", "stream mapping");

    inputList1.stream()
            .flatMap(data -> Arrays.stream(data.split(" ")))
            .forEach(word -> System.out.println(word));
    System.out.println();
}

List<String>이 있습니다. List<Sgring>을 flatMap을 활용하여 Stream<String> 요소로 대체를 하였습니다.

Arrays.Stream()을 사용하면 Stream<String>이 만들어지고, 이 값이 flatMap을 통해서 하나의 Stream<String>으로 리턴이되겠습니다.

String 스트림을 다시 forEach를 활용하여 출력하는 모습입니다.

flatMapToInt()

public static void main(String[] args) {
    List<String> inputList2 = Arrays.asList("10, 20, 30", "40, 50, 60");

    inputList2.stream()
            .flatMapToInt(data -> {
                String[] strArray = data.split(",");
                int[] intArr = new int[strArray.length];
                for (int i = 0; i < strArray.length; i++) {
                    intArr[i] = Integer.parseInt(strArray[i].trim());
                }
                return Arrays.stream(intArr);
            })
            .forEach(number -> System.out.println(number));
}

List<String>타입을 flatMapToInt()을 활용하여 int 타입의 스트림으로 변경하였습니다.

mapToInt()

public static void main(String[] args) {
    List<Student> studentList = Arrays.asList(
        new Student("홍길동", 10),
        new Student("신용권", 20),
        new Student("유미선", 30)
    );

    studentList.stream()
        .mapToInt(Student::getScore)
        .forEach(score->System.out.println(score));
}

mapToInt를 활용하여 기존의 List<Student> 리스트를 IntStream으로 만들어 주고 있습니다.

flatMap()과 mapTo()의 차이점은 flatMap은 기존의 요소를 평평하게 펴주는 역할을 합니다. 하지만 mapTo()는 새로운 요소로 만들어 줍니다.

asDoubleStream()과 boxed() 메소드

public static void main(String[] args) {
    int[] intArray = {1, 2, 3, 4, 5};

    IntStream intStream = Arrays.stream(intArray);
    intStream.asDoubleStream().forEach(d->System.out.println(d));

    System.out.println();

    intStream = Arrays.stream(intArray);
    intStream.boxed().forEach(obj->System.out.println(obj.intValue()));
}

asDoubleStream()을 활용하여 int 타입을 double타입의 DoubleStream을 생성하였습니다.

boxed() 메소드를 활용하여 Stream<Integer> 타입을 생성.

참고 책)

이것이 자바다 (신용권의 Java 프로그래밍 정복)