[Apache Hudi] ExternalSpillableMap: 가용 메모리보다 큰 맵 처리하기
Scalability · 하이브리드 스토리지로 메모리 한계를 넘어 레코드 병합하기
수백만 개의 레코드 업데이트 병합 시 키 매칭에 사용되는 맵이 가용 메모리를 초과하면 프로세스가 실패한다. Apache Hudi는 메모리 부족 시 자동으로 디스크로 스필(spill)하는 하이브리드 자료 구조를 통해 이 문제를 해결한다.
문제: 업데이트 시 레코드 추적하기
데이터 레이크 파일은 불변하기 때문에 업데이트는 별도의 로그 파일로 저장된다. 로그를 베이스 파일과 병합해야 최종 상태에 이른다.
병합 시 레코드 버전을 새 버전으로 교체하기 위해서는 모든 키를 추적해야 한다. 이를 위해 로그의 모든 키를 매핑해야 한다.
대용량 데이터셋에서는 이 맵의 크기가 Executor 메모리를 초과하여 Hudi의 컴팩션과 읽기 작업에서 이 문제가 발생할 수 있다.
단순한 해결책들은 다음과 같은 한계가 있다.
메모리만 사용: 대용량 데이터셋에서 OOM 발생.
디스크만 사용: 랜덤 액세스가 느림.
크기가 고정된 캐시: 데이터 손실 또는 손상 위험.
Hudi의 ExternalSpillableMap은 임계값에 도달하기 전까지 데이터를 메모리에 유지하다가, 초과 시 디스크로 스필한다.
작동 원리
ExternalSpillableMap은 빠른 인메모리(in-memory) HashMap과 지연 초기화(lazily initalized)되는 DiskMap 이렇게 맵 두개를 관리한다.
ExternalSpillableMap.java#L59-L90
maxInMemorySizeInBytes의 20%(SIZING_FACTOR_FOR_IN_MEMORY_MAP의 값)는 JVM 오버헤드를 위해 리저브된다. 예를 들어 1GB의 메모리에 대해 800MB를 사용한 후부터 스필이 시작된다.
Put 연산: 스필 결정
put()은 사용량에 따라 레코드를 메모리 또는 디스크로 라우팅한다.
ExternalSpillableMap.java#L218-L248
아래는 코드를 시각화한 그림이다.
세 가지 경우를 고려해야 한다.
키가 메모리에 있는 경우: 인메모리 맵에서 직접 업데이트.
임계값 미만인 경우: 페이로드 크기 추정 후 메모리에 추가 (이전에 디스크로 스필된 키라면 디스크에서 제거).
임계값 초과인 경우: 디스크 맵을 지연 초기화(lazy initialization)하고 디스크에 기록.
Get과 Iterator 연산
get()과 Iterator 연산은 저장소의 세부 구현을 추상화한다. get()은 먼저 메모리를 확인한 후, 키가 없으면 디스크를 확인한다.
ExternalSpillableMap.java#L208-L216
Iteration은 통합된 뷰를 제공하며, 인메모리 레코드를 소진한 후 디스크로 전환한다.
ExternalSpillableMap.java#L132-L135
사용자는 데이터가 메모리에 있는지 디스크에 있는지 신경 쓰지 않고 순회(iteration)할 수 있다.
적응형 크기 추정
Hudi는 지수 이동 평균을 사용해 레코드 크기를 추정하며, 100개 레코드마다 재계산한다.
90:10 가중치로 최근 레코드 크기에 더 큰 비중을 둔다. 이러한 적응형 방식은 수동 튜닝 없이도 조기 스필이나 OOM 오류를 방지하는 데 도움을 준다.
디스크 백엔드 두 가지: BitCask와 RocksDB
스필된 레코드를 저장할 백엔드로 BitCask 또는 RocksDB를 사용할 수 있으며, 이는 hoodie.common.spillable.diskmap.type를 통해 설정할 수 있다.
BitCask
인메모리 오프셋(offset) 맵을 사용하는 단순한 추가 전용(Append-only) 파일 구조다.
아래는 키와 값이 디스크에 저장되는 방식에 대한 그림이다.
읽기 작업 시에는 RandomAccessFile을 사용해 해당 위치로 직접 이동(seek)하여 데이터를 읽는다.
RocksDB: LSM 트리
RocksDbDiskMap은 쓰기에 최적화된 LSM 트리 저장소인 RocksDB를 사용한다.
RocksDB는 자체적인 컴팩션과 오프힙(off-heap) 메모리 관리를 제공하므로, 대용량 데이터셋에서 더 나은 확장성을 보여준다. BitCask는 모든 키를 메모리(오프셋 맵)에 유지하는 반면, RocksDB는 키를 SSTable에 디스크 저장하기 때문에 키 집합이 RAM보다 커도 처리할 수 있다.
ExternalSpillableMap 사용: HoodieMergedLogRecordScanner
HoodieMergedLogRecordScanner는 컴팩션 및 쿼리 수행 중 레코드를 병합할 때 ExternalSpillableMap을 사용한다.
초기화
스캐너는 메모리 제한과 백엔드 설정으로 맵을 초기화한다.
HoodieMergedLogRecordScanner.java#L120-L123
레코드 처리
processNextRecord()는 중복된 레코드를 병합하는데, 이때 ExternalSpillableMap이 데이터 저장 위치(메모리 vs 디스크)를 추상화하기 때문에 신경쓰지 않아도 된다.
HoodieMergedLogRecordScanner.java#L255-L281
Iteration과 통계
스캐너는 유용한 진단 정보도 로깅한다.
HoodieMergedLogRecordScanner.java#L224-L227
항목 수가 많아 스필이 빈번하여 성능이 떨어진다면 hoodie.memory.merge.fraction을 늘려 스필 발생을 줄일 수 있다.
ExternalSpillableMap의 다른 용도
SpillableMapBasedFileSystemView: 테이블 파일 시스템 뷰를 위한 파일 그룹 메타데이터 캐싱.HoodieCDCLogger: CDC 레코드 버퍼링.
주요 기여자
PR #289 Added support for Disk Spillable Compaction to prevent OOM issues by @n3nash (Nishith Agarwal)
PR #3194 [HUDI-2028] Implement RockDbBasedMap as an alternate to DiskBasedMap by @rmahindra123 (Rajesh Mahindra)