TIL

NoSQL 읽고 쓰기

Document DB

MonogoDB 개요

• MonogoDB 핵심 구조
  • 도큐먼트(Document)
    • 기본 데이터 단위
    • BSON 형식
  • 컬렉션 (Collection)
    • 도큐먼트의 그룹
    • RDBMS 테이블과 유사
  • 데이터베이스 (Database)
    • 컬렉션들의 논리적인 그룹인 최상위 컨테이너
    • RDBMS의 데이터베이스와 유사
• MongoDB의 특징
  • 스키마 정의 없이 유연하기에 요구사항 변화에 빠르게 대응할 수 있다.
  • 수평 확장이 쉬운 분산형 구조이기에 샤딩을 통한 데이터 분산에 용이
  • 대량 데이터를 빠르게 처리할 수 있다.

MongoCursorItemReader

• Spring Batch 5.1에 등장한 MongoDB 기반 커서 ItemReader
• Jdbc의 커서처럼 필요한 만큼한 쿼리 결과를 메모리에 적재한다.
  • MongoDB는 첫 쿼리 실행 시 101개의 도큐먼트를 파견한다.
  • 첫 101개 모두 처리 후 추가 요청 시 batchSize 설정에 따라 다음 데이터가 조회된다.
  • MongoDB는 단일 배치 크기를 16MB로 제한한다. 이는 BSON 도큐먼트 최대 크기가 16MB이기 때문
• MongoCursorItemReader의 최적화
  • 내부 버퍼링
    • 커서에서 받아온 데이터를 내부 버퍼에 임시 보관
  • 최적화된 데이터 접근
    • read() 메서드 호출 시 내부 버퍼를 활용해 비어있을 때만 DB에 요청
    • batchSize만큼의 도큐먼트를 확보

MongoCursorItemReader 해부

MongoCursorItemReader 
  │ 
  ├────── MongoTemplate 
  │         └─ (MongoDB 작업의 핵심 엔진)
  │ 
  ├────── Query (org.springframework.data.mongodb.core.query.Query) 
  │         └─ (MongoDB 쿼리 조건 정의) 
  │
  └────── Cursor 
             └─ (MongoDB 데이터를 순차적으로 읽어오는 스트리밍 객체)

• MongoTemplate
  • Spring의 MongoTemplate으로 데이터를 스트리밍으로 조회한다.
• Query
  • 데이터를 조회할 쿼리를 지정
  • doOpen()에 최초로 쿼리를 생성, 커서를 얻는다.
• Cursor
  • read() 메서드가 호출될 때마다 데이터를 순차적으로 반환
• MongoCursorItemReader 처리 절차
  • 버퍼에 데이터가 있는 경우
    • read() 호출
    • 버퍼의 다음 Document 반환
  • 버퍼에 데이터가 없는 경우
    • read() 호출
    • 새 Document 요청 → Document 일괄 전송 (batchSize 단위)
    • 버퍼의 다음 Document 반환

@Bean
@StepScope
public MongoCursorItemReader<SecurityLog> securityLogReader(
        @Value("#{jobParameters['searchDate']}") LocalDate searchDate
) {
    Date startOfDay = Date.from(searchDate.atStartOfDay(ZoneId.systemDefault()).toInstant());
    Date endOfDay = Date.from(searchDate.plusDays(1).atStartOfDay(ZoneId.systemDefault()).toInstant());

    return new MongoCursorItemReaderBuilder<SecurityLog>()
            .name("securityLogReader")
            .template(mongoTemplate)
            .collection("security_logs")
            .jsonQuery("""
            {
                "label": "PENDING_ANALYSIS",
                "timestamp": {
                    "$gte": ?0,
                    "$lt": ?1
                }
            }
            """)
            .parameterValues(List.of(startOfDay, endOfDay))
            .sorts(Map.of("timestamp", Sort.Direction.ASC))
            .targetType(SecurityLog.class)
            .batchSize(10)
            .build();
}