Spring Batch 구현 - 6

 
♧ 전체 코드 : https://github.com/woodisco/pass-batch

GitHub - woodisco/pass-batch

 

통계 데이터 생성 배치

① StatisticsRepository 작성
② StatisticsEntity, AggregatedStatistics 작성
③ CustomCSVWriter 작성
④ MakeDailyStatisticsTasklet 작성
⑤ MakeWeeklyStatisticsTasklet 작성
⑥ MakeStatisticsJobConfig 작성

@Slf4j
@Configuration
public class MakeStatisticsJobConfig {
    private final int CHUNK_SIZE = 5;

    private final EntityManagerFactory entityManagerFactory;

    private final StatisticsRepository statisticsRepository;

    private final MakeDailyStatisticsTasklet makeDailyStatisticsTasklet;

    private final MakeWeeklyStatisticsTasklet makeWeeklyStatisticsTasklet;

    public MakeStatisticsJobConfig(EntityManagerFactory entityManagerFactory, StatisticsRepository statisticsRepository, MakeDailyStatisticsTasklet makeDailyStatisticsTasklet, MakeWeeklyStatisticsTasklet makeWeeklyStatisticsTasklet) {
        this.entityManagerFactory = entityManagerFactory;
        this.statisticsRepository = statisticsRepository;
        this.makeDailyStatisticsTasklet = makeDailyStatisticsTasklet;
        this.makeWeeklyStatisticsTasklet = makeWeeklyStatisticsTasklet;
    }

    @Bean
    public Job makeStatisticsJob(JobRepository jobRepository,
                                 Step addStatisticsStep,
                                 Step makeDailyStatisticsStep,
                                 Step makeWeeklyStatisticsStep) {
        Flow addStatisticsFlow = new FlowBuilder<Flow>("addStatisticsFlow")
                .start(addStatisticsStep)
                .build();
        Flow makeDailyStatisticsFlow = new FlowBuilder<Flow>("makeDailyStatisticsFlow")
                .start(makeDailyStatisticsStep)
                .build();

        Flow makeWeeklyStatisticsFlow = new FlowBuilder<Flow>("makeWeeklyStatisticsFlow")
                .start(makeWeeklyStatisticsStep)
                .build();

        Flow parallelMakeStatisticsFlow = new FlowBuilder<Flow>("parallelMakeStatisticsFlow")
                .split(new SimpleAsyncTaskExecutor())
                .add(makeDailyStatisticsFlow, makeWeeklyStatisticsFlow)
                .build();

        return new JobBuilder("makeStatisticsJob", jobRepository)
                .start(addStatisticsFlow)
                .next(parallelMakeStatisticsFlow)
                .build()
                .build();
    }

    @Bean
    public Step addStatisticsStep(JobRepository jobRepository, PlatformTransactionManager transactionManager) {
        return new StepBuilder("addStatisticsStep", jobRepository)
                .<BookingEntity, BookingEntity>chunk(CHUNK_SIZE, transactionManager)
                .reader(addStatisticsItemReader(null, null))
                .writer(addStatisticsItemWriter())
                .build();
    }

    @Bean
    @StepScope
    public JpaCursorItemReader addStatisticsItemReader(@Value("#{jobParameters[from]}") String fromString, @Value("#{jobParameters[to]}") String toString) {
        final LocalDateTime from = LocalDateTimeUtils.parse(fromString);
        final LocalDateTime to = LocalDateTimeUtils.parse(toString);

        return new JpaCursorItemReaderBuilder<BookingEntity>()
                .name("usePassesItemReader")
                .entityManagerFactory(entityManagerFactory)
                // JobParameter를 받아 종료 일시(endedAt) 기준으로 통계 대상 예약(Booking)을 조회합니다.
                .queryString("select b from BookingEntity b where b.endedAt between :from and :to")
                .parameterValues(Map.of("from", from, "to", to))
                .build();
    }

    @Bean
    public ItemWriter<BookingEntity> addStatisticsItemWriter() {
        return bookingEntities -> {
            Map<LocalDateTime, StatisticsEntity> statisticsEntityMap = new LinkedHashMap<>(); // 순서대로 나열

            for (BookingEntity bookingEntity : bookingEntities) {
                final LocalDateTime statisticsAt = bookingEntity.getStatisticsAt();
                StatisticsEntity statisticsEntity = statisticsEntityMap.get(statisticsAt);

                if(statisticsEntityMap.isEmpty()) {
                    statisticsEntityMap.put(statisticsAt, StatisticsEntity.create(bookingEntity));
                } else {
                    statisticsEntity.add(bookingEntity);
                }
            }
            final List<StatisticsEntity> statisticsEntities = new ArrayList<>(statisticsEntityMap.values());
            statisticsRepository.saveAll(statisticsEntities);
        };
    }

    @Bean
    public Step makeDailyStatisticsStep(JobRepository jobRepository, PlatformTransactionManager transactionManager) {
        return new StepBuilder("makeDailyStatisticsStep", jobRepository)
                .tasklet(makeDailyStatisticsTasklet, transactionManager)
                .build();
    }

    @Bean
    public Step makeWeeklyStatisticsStep(JobRepository jobRepository, PlatformTransactionManager transactionManager) {
        return new StepBuilder("makeWeeklyStatisticsStep", jobRepository)
                .tasklet(makeWeeklyStatisticsTasklet, transactionManager)
                .build();
    }
}

▷ Flow : 
Spring Batch에서 작업을 순차적으로 조직하는 데 사용되는 구성 요소입니다. Flow는 단계(Step)의 컬렉션을 그룹화하고, 이러한 단계들 간의 제어 흐름을 정의하는 데 사용됩니다. 일반적으로 Flow는 작업(Job)의 실행 흐름을 설명하고, 여러 단계로 구성된 복잡한 처리 과정을 구성하는 데 사용됩니다.
split 메서드를 사용하여 병렬 처리를 지정하고, add 메서드를 사용하여 각 Flow를 추가합니다.

▷ @StepScope :
이 어노테이션은 스텝 단위로 빈(bean)을 생성하고 관리하는 데 사용됩니다.
일반적으로 Spring Batch 작업(Job)은 여러 단계(Step)로 구성됩니다. 각 단계는 데이터 처리의 특정 부분을 담당하고 있으며, 각 단계는 별도의 트랜잭션을 가질 수 있습니다. 이러한 특성으로 인해 스텝 간의 데이터 공유나 빈의 범위(scope) 설정이 중요해집니다.
@StepScope 어노테이션은 이러한 상황에서 스텝 단위로 빈을 생성하고 스텝의 콘텍스트에 해당 빈을 등록합니다. 이를 통해 각 스텝이 실행될 때마다 새로운 빈 인스턴스가 생성되며, 스텝 간의 데이터 공유를 효과적으로 방지할 수 있습니다.
예를 들어, Spring Batch에서는 ItemReader, ItemProcessor, ItemWriter와 같은 컴포넌트를 사용하여 데이터 처리를 수행합니다. 이러한 컴포넌트 중 하나가 스텝 단위로 빈이 생성되어야 할 때 @StepScope 어노테이션을 사용할 수 있습니다. 이렇게 하면 각 스텝이 실행될 때마다 해당 빈이 새로 생성되어 데이터 처리 도중에 발생할 수 있는 문제를 방지할 수 있습니다.

간단히 말해, @StepScope 어노테이션은 Spring Batch에서 스텝 단위로 빈을 생성하고 관리하여 각 스텝이 독립적으로 동작하고 데이터를 안전하게 처리할 수 있도록 도와줍니다.

▷ jobParameters :
Spring Batch에서 작업(Job) 실행 시 전달되는 매개변수들을 포함하는 컨테이너입니다. 이러한 매개변수는 Spring Batch 작업을 실행하는 데 필요한 추가 정보를 제공하고, 작업의 동작을 제어하는 데 사용됩니다.

⑦ LocalDateTimeUtils에 메소드 추가
 

출처 : 패스트캠퍼스 10개 프로젝트로 완성하는 백엔드 웹개발(Java/Spring) 초격차 패키지 Online