TIL

NoSQL 읽고 쓰기 - MongoDB

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();
}

MongoPagingItemReader

• 페이징 방식으로 데이터를 조회하는 ItemReader
• skip()과 limit() 연산자를 사용
• offset 기반의 페이징을 처리
  • 때문에 성능이 좋지 않다.
• 샤딩 환경에서 특히 더 성능이 좋지 않다.
  • 각 샤드에서 데이터에 대해 정렬과 offset만큼의 skip이 발생
  • 각 샤드에서 받은 모든 데이터를 모두 모아 다시 정렬한 뒤 전체 결과에서 offset을 적용
• 때문에 실무 환경에서 사용하지 않는 것을 추천한다.
  • 페이징이 필요하다면 keyset 기반 페이징을 직접 만들어 수행하는 것을 권장

MongoPagingItemReader 구성 요소

• MongoCursorItemReader와 달리 매 페이지마다 새로운 쿼리를 실행한다.

@Bean
@StepScope
public MongoPagingItemReader<SecurityLog> securityLogReader(
    @Value("#{jobParameters['searchDate']}") LocalDate searchDate
) {
    Query query = new Query()
            .addCriteria(Criteria.where("label").is("PENDING_ANALYSIS"))
            .addCriteria(Criteria.where("timestamp")
                    .gte(Date.from(searchDate.atStartOfDay(ZoneId.systemDefault()).toInstant()))
                    .lt(Date.from(searchDate.plusDays(1).atStartOfDay(ZoneId.systemDefault()).toInstant())))
            .with(Sort.by(Sort.Direction.ASC, "timestamp"));

    return new MongoPagingItemReaderBuilder<SecurityLog>()
            .name("securityLogReader")
            .template(mongoTemplate)
            .collection("security_logs")
            .query(query)
            .sorts(Map.of("timestamp", Sort.Direction.ASC))  // 빌더의 버그 때문에 필요
            .targetType(SecurityLog.class)
            .pageSize(10)
            .build();
}

MongoItemWriter

• MongoTemplate을 사용해 청크의 아이템을 추가/수정/삭제한다.

bulkWrite

• 몽고의 bulkWrite는 JDBC의 batchUpdate와 유사한 컨셉으로 작동한다.
• 아래처럼 여러 쓰기 작업을 하나의 배열로 모아 한 번의 네트워크 요청으로 처리한다.
  • 순차적으로 실행
  • 하나가 실패하면 후속 작업은 실행되지 않는다.
  • 청크마다 bulkWrite를 호출한다.

// bulkWrite에는 몽고DB에서 지원하는 다양한 쓰기 연산을 사용할 수 있다. 
// 하지만 MongoItemWriter는 아래 세 가지 연산만 사용한다.
db.collection.bulkWrite(
   [
      { insertOne : <document> },
      { replaceOne : <document> },
      { deleteMany : <document> }
   ]
)

MongoItemWriter 해부

MongoItemWriter
│
├────── MongoTemplate
│       └─ (MongoDB 작업을 수행하는 핵심 컴포넌트)
│
├────── collection
│       └─ (데이터를 저장할 MongoDB 컬렉션명)
│
└────── mode
        └─ (INSERT/UPSERT/REMOVE - 도큐먼트 처리 방식 - 쓰기 연산 결정에 사용)

• mode - MongoItemWriter가 도큐먼트를 어떻게 처리할지 결정
  • INSERT: 도큐먼트 추가 (insertOne)
  • UPSERT: 기존 도큐먼트 수정, 존재하지 않으면 추가 (기본값) (replaceOne)
  • REMOVE: 도큐먼트 삭제 (deleteMany)

MongoTransactionManager

• bulkWrite 도중 하나의 작업이 실패하면 후속 작업들을 실행되지 않는다.
  • 문제는 앞선 작업들이 롤백되지 않는다는 것이다.
• MongoTransactionManaager
  • 스텝의 트랜잭션 매니저로 별도 구성하면 롤백이 가능하다.
  • 다만 다음 조건이 필요하다.
    • Mongo유 4.0+
    • 레플리카 셋 또는 샤드 클러스터 환경에서만

private final PlatformTransactionManager mongoTransactionManager;

public HackerPatternDetectionJob(
        JobRepository jobRepository,
        MongoTemplate mongoTemplate) 
{
    ...
    this.mongoTransactionManager = new MongoTransactionManager(mongoTemplate.getMongoDatabaseFactory());
}

...
// Step의 트랜잭션 매니저로 MongoTransactionManager 사용
<SecurityLog, SecurityLog>chunk(10, mongoTransactionManager)