TIL

FlatFileItemReader

파일 기반 배치 처리

• 파일 기반 데이터 처리는 실무에서 여전히 많이 쓰이고 중요한 역할을 한다.
  • ex) CSV, XML, JSON 등

파일 처리의 실체

• 파일 처리를 직접 구현한다면?
  • 파일을 열고 닫는 것부터가 힘들다.
  • 대용랑 파일의 경우 OutOfMemoryError가 나기 십상이다
  • try-catch로 점칠된 예외 처리에 스파게티 코드가 되기 쉽다.
  • 멀티스레드 환경이라면 데드락과 레이스 컨디션 확률이 높다.
  • 청크 일부만 쓰다 실패한다면 트랜잭션 롤백은 직접 해야 한다.

파일 기반 배치 처리

• 가장 기본적이면서 강력한 도구 두 가지는 다음과 같다.
  • FlatFileItemReader
  • FlatFileItemWriter

Flat 파일이란

• CSV 같은 파일을 싱각하면 된다.
  • 단순히 열과 행으로만 구성된 파일
• 플랫 파일의 특징
  • 각 라인이 하나의 데이터 (레코드 개념)
  • 다양한 구분자 지원
    • CSV의 경우 쉼표로 각 필드를 구분한다.
    • TSV의 경우 탭(\t)으로 구분한다.
  • 강력한 호환성과 범용성
    • 거의 모든 시스템에서 읽고 쓸 수 있는 표준이다.

FlatFileItemReader

• 플랫 파일로부터 데이터를 읽어오는 Spring Batch의 핵심 도구 중 하나
• read() 메서드 동작은 크게 두 단계로 이루어진다.
  • 파일에서 한 줄을을 읽기
  • 읽어온 한 줄을 객체로 변환해 리턴

// FlatFileItemReader.doRead()
// ...
String line = readLine();
// ...
return lineMapper.mapLine(line, lineCount); 

• 객체 변환 과정에선 LineMapper라는 컴포넌트가 핵심을 담당한다.
  • Spring JDBC의 RowMapper와 유사

public interface LineMapper<T> {
    T mapLine(String line, int lineNumber) throws Exception;
}

DefaultLineMapper

• Spring Batch는 이미 강력한 기본 구현체인 DefaultLineMapper를 제공한다.
• DefaultLineMapper 동작도 크게 두 단계로 이루어진다.
  • 토근화(Tokenization)
    • 하나의 문자열 라인을 토큰 단위로 분리
    • 구분자를 기준으로 각 컬럼을 분리한다.
  • 객체 매핑
    • 분리된 토큰들을 객체 프로퍼티에 매핑

@Override
public T mapLine(String line, int lineNumber) throws Exception {
    FieldSet fieldSet = tokenizer.tokenize(line);  // 1단계: 토큰화
    return fieldSetMapper.mapFieldSet(fieldSet);  // 2단계: 객체 매핑	 
}

1단계: 토큰화 - LineTokenizer

• 토큰화 단계에선 LineTokenizer 컴포넌트가 사용된다.
• LineTokenizer 구현체는 대표적으로 다음 두 가지가 있다.
  • DelimitedLineTokenizer - 쉼표(,) 구분자로 데이터를 토큰화
    • ex) ERR001,2024-01-19,CRITICAL -> ["ERR001", "2024-01-19", "CRITICAL"]
  • FixedLengthTokenizer - 고정된 길이로 구분된 데이터를 토큰화
    • ERR00120240119CRITICAL -> ["ERR001", "20240119", "CRITICAL"] (각 6, 8, 8자리)
  • 어떤 구현체를 사용할지 FlatFileItemReader 구성 시점에 선택 가능하다.
• 데이터를 토큰화한 결과를 FieldSet 객체로 만들어 변환한다.

public class DefaultFieldSet implements FieldSet {
   private final String[] tokens; // 토큰화된 데이터
   private List<String> names; // 각 데이터를 객체의 어떤 프로퍼티에 매핑할지 나타내는 프로퍼티 이름 목록
   // ...
}

• ex) "ERR001,2024-01-19 10:15:23,CRITICAL,1234,SYSTEM_CRASH”
  • tokens: ["ERR001", "2024-01-19 10:15:23", "CRITICAL", "1234", "SYSTEM_CRASH"]
  • names: ["errorId", "errorDateTime", "severity", "processId", "errorMessage"]

2단계: 객체 매핑 - FieldSetMapper

• 객체 매핑 단계에선 FieldSetMapper가 사용된다.

public interface FieldSetMapper<T> {
    T mapFieldSet(FieldSet fieldSet) throws BindException;
}

• 별도 설정이 없다면 기본적으로 BeanWrapperFieldSetMapper가 사용된다.
  • 자바 빈 규약을 따르는 구현체
  • 객체의 setter 메서드를 호출해서 데이터를 설정한다.
  • FieldSet 필드 이름과 객체 프로퍼티 이름이 일치해야 한다.

구분자로 분리된 형식의 파일 읽기

Step 구성

@Bean
fun systemFailureStep(
    systemFailureItemReader: FlatFileItemReader<SystemFailure>,
    systemFailureStdoutItemWriter: SystemFailureStdoutItemWriter,
): Step {
    return StepBuilder("systemFailureStep", jobRepository)
        .chunk<SystemFailure, SystemFailure>(10, transactionManager)
        .reader(systemFailureItemReader)
        .writer(systemFailureStdoutItemWriter)
        .build()
}

• chunk의 제네릭 타입을 SystemFailure, 즉 변환할 객체로 설정한다.

// 기본 생성자가 필요하여 모두 기본값을 설정
// setter가 필요하여 var로 선언
data class SystemFailure(
    var errorId: String = "",
    var errorDateTime: String = "",
    var severity: String = "",
    var processId: Int = 0,
    var errorMessage: String = "",
)

FlatFileItemReader 구성 해부

• FlatFileItemReaderBuilder라는 전용 빌더 클래스를 사용해 구성

@Bean
@StepScope
fun systemFailureItemReader(
    @Value("#{jobParameters['inputFile']}") inputFile: String,
): FlatFileItemReader<SystemFailure> {
    return FlatFileItemReaderBuilder<SystemFailure>()
        .name("systemFailureItemReader")
        .resource(ClassPathResource(inputFile))
        .delimited()
        .delimiter(",")
        .names(
            "errorId",
            "errorDateTime",
            "severity",
            "processId",
            "errorMessage"
        )
        .targetType(SystemFailure::class.java)
        .linesToSkip(1)
        .build()
}

• name(): ItemReader 식별자 지정
  • 고유 이름을 부여하여 진행 상황 등을 추적할 때 사용
• resource(): 읽어 들일 Resource를 지정
• delimited() 파일 형식 지정
  • 구분자로 분리된 형식임을 알리는 설정
  • delimited를 호출하면 DefaultLineMapper가 사용되어 DelimitedLineTokenizer가 지정된다.
• delimiter(): 구분자 지정
  • 기본 구분자가 쉼표이기에 CSV 처리 시엔 생략해도 된다.
  • 하지만 명시성을 위해 직접 지정하는 것이 권장된다.
• names(): 프로퍼티 매핑
• targetTypes(): 매핑 대상 클래스 지정
  • 기본적으로 FieldSetMapper 구현체인 BeanWrapperFieldSetMapper가 사용된다.
  • 제네릭은 런타임에 소거되기에 BeanWrapperFieldSetMapper가 새 인스턴스를 생성하려면 tartetType 설정이 필요하다.
• linesToSkip(): 헤더 처리
  • 보통 첫 줄은 컬럼명으로 사용하기에 스킵한다. (linesToSkip(1))
• strict(): 파일 검증 강도 설정
  • 기본값은 true로 파일 누락 시 배치를 중단한다.
  • false면 파일이 없어도 경고만 남기고 진행한다.
    • read() 메서드에서 null 반환

고정 길이 형식 파일 읽기

• 고절 길이 형식 파일은 각 필드가 고정된 길이로 맞춰진 텍스트 파일이다.
  • 주로 레거시 시스템에서 사용

@Bean
@StepScope
fun systemFailureItemReader(
    @Value("#{jobParameters['inputFile']}") inputFile: String,
): FlatFileItemReader<SystemFailure> {
    return FlatFileItemReaderBuilder<SystemFailure>()
        .name("systemFailureItemReader")
        .resource(ClassPathResource(inputFile))
        .fixedLength()
        .columns(
            Range(1, 8),  // errorId: ERR001 + 공백 2칸
            Range(9, 29),  // errorDateTime: 날짜시간 + 공백 2칸
            Range(30, 39),  // severity: CRITICAL/FATAL + 패딩
            Range(40, 45),  // processId: 1234 + 공백 2칸
            Range(46, 66) // errorMessage: 메시지 + \n
        )
        .names("errorId", "errorDateTime", "severity", "processId", "errorMessage")
        .targetType(SystemFailure::class.java)
        .build()
}

• fixedLength(): 고정 길이 형식임을 알리는 설정
  • LineTokenizer 구현체로 FixedLengthTokenizer가 지정된다.
• columns(): Range 배열로 각 필드의 시작과 끝을 정의
• strict()
  • 파일 존재 여부, 토큰 개수 불일치 외에도 라인 길이 검증의 엄격도를 설정한다.
  • Range에 지정된 최대 길이가 다를 경우 예외를 발생시킨다.
• names(), targetType()
  • 구분자 형식 파일 처리와 동일하게 동작한다.

프로퍼티 타입에 LocalDateTime 사용

• BeanWrapperFieldSetMapper은 기본 타입 외엔 변환을 지원하지 않는다.
• customEditors() 메서드로 커스텀 PropertyEditor를 등록할 수 있다.

@Bean
@StepScope
fun systemFailureItemReader(
    @Value("#{jobParameters['inputFile']}") inputFile: String,
): FlatFileItemReader<SystemFailure> {
    return FlatFileItemReaderBuilder<SystemFailure>()
        // ...
        .customEditors(mapOf(LocalDateTime::class.java to dateTimeEditor()))
        .build()
        
private fun dateTimeEditor(): PropertyEditor {
    return object : PropertyEditorSupport() {
        override fun setAsText(text: String) {
            val formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")
            setValue(LocalDateTime.parse(text, formatter))
        }
    }
}       

• 파일에서 읽은 날짜 문자열이 자동으로 LocalDateTime 객체로 변환된다.

RegexLineTokenizer

• 구분자 기반 형식과 고정 길이 형식은 복잡한 형태의 데이터는 처리하지 못한다.
  • ex)[WARNING][Thread-156][CPU: 78%] Thread pool saturation detected - 45/50 threads in use…
  • 로그에서 Thread 번호나 에러 메시지 등 특정 데이터를 추출하려면 구분자나 고정 길이로는 추출할 수 없다.

RegexLineTokenizer 구성

• RegexLineTokenizer는 정규식을 활용한 토큰 파싱 도구다.

@Bean
@StepScope
fun logItemReader(
    @Value("#{jobParameters['inputFile']}") inputFile: String,
): FlatFileItemReader<LogEntry> =
    FlatFileItemReaderBuilder<LogEntry>()
        .name("logItemReader")
        .resource(ClassPathResource(inputFile))
        .lineTokenizer(
            RegexLineTokenizer().apply {
                setRegex("\\[\\w+\\]\\[Thread-(\\d+)\\]\\[CPU: \\d+%\\] (.+)")
            }
        )
        .fieldSetMapper { fieldSet: FieldSet ->
            LogEntry(
                fieldSet.readString(0),
                fieldSet.readString(1)
            )
        }
        .build()

data class LogEntry(
    var threadNum: String = "",
    var message: String = "",
)    

• lineTokenizer 등록
  • 정규식을 세팅한 뒤 builder의 .lineTokenizer()로 구성해주면 등록된다.
• 커스텀 filedSetMapper 등록
  • FlatFileItemReaderBuilder는 커스텀 filedSetMapper도 등록할 수 있다.
  • 커스텀 매퍼를 사용할 땐 .targetType() 메서드를 호출하면 안 된다.

PatternMatchingCompositeLineMapper

• RegexLineTokenizer는 하나의 형식만 지원하기에 파일의 모든 라인이 해당 정규식을 따르는 경우 유용하다.
  • 하지만 로그는 보통 로그 레벨에 따라 다른 형식일 수도 있다.
• PatternMatchingCompositeLineMapper
  • Ant 스타일 패턴 매칭을 지원하여 각 라인의 패턴에 따라 서로 다른 LineTokenizer와 FieldSetMapper를 적용할 수 있다.

  유형별 SystemLog 구현체 예시

  • 각 로그 타입별로 아래와 같은 형식을 지원한다고 가정하자

    // ERROR,mysql-prod,OOM,2024-01-24T09:30:00,heap space killing spree,85%,/var/log/mysql
    // ABORT,spring-batch,MemoryLeak,2024-01-24T10:15:30,forced termination,-1,/usr/apps/batch,TERMINATED
    // COLLECT,heap-dump,PID-9012,2024-01-24T11:00:15,/tmp/heapdump
        
    sealed interface SystemLog {
        val type: String
        val timestamp: String
        
        data class Error(
            override val type: String,
            override val timestamp: String,
            val application: String,
            val errorType: String,
            val message: String,
            val resourceUsage: String,
            val logPath: String,
        ) : SystemLog
        
        data class Abort(
            override val type: String,
            override val timestamp: String,
            val application: String,
            val errorType: String,
            val message: String,
            val exitCode: String,
            val processPath: String,
            val status: String,
        ) : SystemLog
        
        data class Collect(
            override val type: String,
            override val timestamp: String,
            val dumpType: String,
            val processId: String,
            val dumpPath: String,
        ) : SystemLog
    }
    

LineMapper

• LineTokenizer, FieldSetMapper를 한 번에 설정한 LineMapper를 설정할 수 있다.

    @Bean
    @StepScope
    fun systemLogReader(
        @Value("#{jobParameters['inputFile']}") inputFile: String,
    ): FlatFileItemReader<SystemLog> {
        return FlatFileItemReaderBuilder<SystemLog>()
            // ...
            .lineMapper(systemLogLineMapper())
            .build()
    }

systemLogLineMapper 정의

• systemLogLineMapper 빈 정의는 아래와 같다.
  • setTokenizers, setFieldSetMappers을 통해 각 패턴별 필요한 구현체들을 설정한다.

@Bean
fun systemLogLineMapper(): PatternMatchingCompositeLineMapper<SystemLog> =
    PatternMatchingCompositeLineMapper<SystemLog>()
        .apply {
            setTokenizers(
                mapOf(
                    "ERROR*" to errorLineTokenizer(),
                    "ABORT*" to abortLineTokenizer(),
                    "COLLECT*" to collectLineTokenizer(),
                )
            )
            setFieldSetMappers(
                mapOf(
                    "ERROR*" to ErrorFieldSetMapper(),
                    "ABORT*" to AbortFieldSetMapper(),
                    "COLLECT*" to CollectFieldSetMapper(),
                )
            )
        }

@Bean
fun errorLineTokenizer(): DelimitedLineTokenizer =
    DelimitedLineTokenizer(",").apply {
        setNames("type", "application", "errorType", "timestamp", "message", "resourceUsage", "logPath")
    }
// ...

class ErrorFieldSetMapper : FieldSetMapper<SystemLog> {
    override fun mapFieldSet(fieldSet: FieldSet): SystemLog =
        SystemLog.Error(
            type = fieldSet.readString("type"),
            timestamp = fieldSet.readString("timestamp"),
            application = fieldSet.readString("application"),
            errorType = fieldSet.readString("errorType"),
            message = fieldSet.readString("message"),
            resourceUsage = fieldSet.readString("resourceUsage"),
            logPath = fieldSet.readString("logPath"),
        )
}

// ...

MultiResourceItemReader: 여러 파일 읽기

• MultiResourceItemReader
  • 여러 파일을 순차적으로 읽을 수 있게 해주는 ItemReader 구현체
  • MultiResourceItemReader가 실제로 파일 읽기를 수행하지 않는다.
  • 실제 파일 읽기는 위임 대상 ItemReader가 수행한다.
  • MultiResourceItemReaderBuilder 전용 빌더를 통해 생성할 수 있다.

    @Bean
    @StepScope
    fun multiSystemFailureItemReader(
        @Value("#{jobParameters['inputFilePath']}") inputFilePath: String,
    ): MultiResourceItemReader<SystemFailure> =
        MultiResourceItemReaderBuilder<SystemFailure>()
            .name("multiSystemFailureItemReader")
            .resources( // 읽을 대상 Resource 목록을 배열 형태로 구성
                ClassPathResource("$inputFilePath/critical-failures.csv"),
                ClassPathResource("$inputFilePath/normal-failures.csv"),
            ).delegate(systemFailureReader()) // 위임 ItemReader 지정
            .build()

• 파일 간의 순서 지정이 필요하다면 comparator를 활용할 수 있다.

.comparator { r1, r2 -> r2.filename!!.compareTo(r1.filename!!) }
.build()

• delegate로 지정한 위임 대상 ItemReader는 다른 파일 읽기 ItemReader와 다르게 resource 정보를 파라미터로 받지 않는다.
  • MultiResourceItemReader가 직접 지정해주기 때문

    @Bean
    fun systemFailureReader(): FlatFileItemReader<SystemFailure> =
        FlatFileItemReaderBuilder<SystemFailure>()
            .name("systemFailureReader")
            // ...