[Kafka] Log Compaction, Message Compression

TL;DR

- 모두 저장 공간 효율을 위해 사용될 수 있는 기능이다.

- Log Compaction은 같은 record key를 갖는 가장 최신 로그만 남기고 삭제하는 것을 말한다.

- Message Compression은 메시지 자체를 압축해서 전달하는 것을 말한다.

 

1. Message Compression (메시지 압축)

- Message Compression은 메시지의 크기를 줄이는 기술

- 메시지 크기를 줄임으로써 네트워크 대역폭을 절약하고 저장 공간을 효율적으로 사용하며, 더 빠른 전송 및 처리 속도를 제공

- 주로 사용되는 압축 알고리즘에는 Gzip, Snappy, LZ4 등이 있음

- 프로듀서 레벨 압축이 있고, 브로커 레벨 압축이 있음

- 프로듀서 레벨 압축은 전송되는 데이터 양을 줄이기 위해 데이터가 전송되기 전에 메시지를 압축하고, 수신 측에서 이를 해제하는 방식으로 동작

 

관련 옵션

compression.type	메시지 압축 알고리즘 (gzip, snappy, lz4, zstd, none)	producer
message.max.bytes	Kafka 브로커가 허용하는 최대 메시지 크기 (바이트)	1000012 (약 1MB)
max.in.flight.requests.per.connection	한 네트워크 연결에서 동시에 허용되는 최대 미완료 요청 수	5
linger.ms	프로듀서가 메시지를 배치로 보낼 때까지 대기하는 시간 (밀리초)	0
batch.size	프로듀서가 한 번에 전송할 수 있는 최대 배치 크기 (바이트)	16384 (16KB)

 

테스트 결과

 

2. Log Compaction (로그 압축)

- Log Compaction은 Apache Kafka와 같은 메시지 브로커 시스템에서 사용되는 기법으로, 로그 데이터의 저장 공간을 최적화하고 불필요한 데이터를 정리하는 데 사용

- Kafka에서 Log Compaction은 동일한 키를 가진 오래된 메시지를 삭제하고 최신 상태만 유지함으로써 토픽의 저장 공간을 절약

- 데이터의 영구 보존이 필요할 때 유용하며, 특정 키의 최신 상태를 필요로 하는 애플리케이션에서 유용

- 이를 통해 데이터의 최신 상태를 유지하면서도 저장 공간을 효율적으로 사용할 수 있습니다.

- Log compaction은 절대 메시지의 순서를 바꾸지 않고 단지 몇개를 삭제한다.  Partition offset 또한 바뀌지 않는다.

- Compaction된 메시지들도 fromBeginning 잘 동작한다.

- 활성 세그먼트는 대상에 포함되지 않는다 세그먼트 롤링은 log.segment.bytes(세그먼트 크기) 또는 log.roll.ms(세그먼트의 최대 수명) 등의 설정도 봐야한다.

 

 

 

관련 옵션

설정 이름 설명 기본값

log.cleaner.enable	Log Compaction을 활성화하는 설정	true
log.cleaner.threads	Log Compaction을 수행하는 쓰레드 수	1
log.cleanup.policy	로그 정리 정책 (compact 또는 delete)	delete
log.cleaner.min.cleanable.ratio	- 로그가 정리 가능한 최소 비율 - 특정 로그 세그먼트에서 정리 가능 부분이 전체 크기의 50% 이상이라면, 이 세그먼트는 컴팩션 후보가 되며, 예를 들어, 1GB의 로그 세그먼트에서 500MB 이상이 오래된 메시지로 간주될 경우 컴팩션이 트리거됩니다.	0.5
delete.retention.ms	삭제된 레코드를 보관하는 시간 (밀리초)	86400000 (24시간)
min.compaction.lag.ms	- 메시지가 컴팩션될 때까지 최소 지연 시간 (밀리초) - 컴팩션이 지연되도록 설정하여 데이터가 너무 일찍 제거되지 않도록 유도	0
max.compaction.lag.ms	- 메시지가 컴팩션될 때까지 최대 지연 시간 (밀리초) - 이 값이 지나면 해당 메시지는 반드시 컴팩션되며, 시스템이 최대한 빨리 공간을 회수하도록 유도	9223372036854775807 (무한대)

 

Kafka Log Cleaner

- Kafka는 log을 가지고 log는 parition으로 나누어지고, partition은 segment로 나누어진다.

- segment는 key, value를 포함하고 있다. Log Cleaner는 log compaction을 수행하고 compaction thread pool을 가진다.

- 이 thread들은 가장 최신 버전 로그들만 선택해서 segment를 구성한후 이전 segment와 바꾼다.

조건 충족 요소

1. log.cleaner.min.cleanable.ratio
   • 설명: 이 설정은 로그 세그먼트가 컴팩션되기 위해 필요한 최소 정리 가능 비율을 나타냅니다. 로그 세그먼트에서 "정리 가능"한 부분은 동일한 키에 대해 더 이상 유효하지 않은 오래된 메시지입니다.
   • 기본값: 0.5
   • 작동 방식: 만약 특정 로그 세그먼트에서 정리 가능 부분이 전체 크기의 50% 이상이라면, 이 세그먼트는 컴팩션 후보가 됩니다. 예를 들어, 1GB의 로그 세그먼트에서 500MB 이상이 오래된 메시지로 간주될 경우 컴팩션이 트리거됩니다.
2. 로그 세그먼트 크기
   • 로그 세그먼트의 크기가 일정 수준을 초과하면 컴팩션이 발생할 수 있습니다. 로그 세그먼트는 Kafka가 로그 데이터를 관리하는 기본 단위로, 세그먼트가 너무 커지면 컴팩션이 필요할 수 있습니다. 이 세그먼트 크기는 log.segment.bytes 설정으로 관리됩니다.
3. 디스크 사용량
   • 디스크 사용량이 일정 수준을 초과하면 Kafka는 컴팩션을 통해 디스크 공간을 회수하려고 합니다. 이는 디스크 용량이 제한적일 때 특히 중요합니다. 디스크 사용량 관리와 관련된 설정에는 log.retention.bytes와 log.retention.ms 등이 있습니다.
4. 메시지 수
   • 특정 파티션에서 관리하는 메시지 수가 증가하면 컴팩션이 발생할 수 있습니다. 메시지 수와 관련된 설정으로는 log.retention.ms와 log.retention.bytes가 있으며, 이들은 메시지의 보존 기간과 크기를 제한하는 데 사용됩니다.
5. 컴팩션 지연 시간
   • min.compaction.lag.ms와 max.compaction.lag.ms 설정은 메시지가 컴팩션되기 전에 경과해야 하는 최소 및 최대 시간을 지정합니다. 이 설정을 통해 메시지가 너무 빨리 컴팩션되지 않도록 하고, 일정 시간 이후에는 컴팩션이 반드시 발생하도록 합니다.

Log Compaction이 실행되는 방식

1. 백그라운드 프로세스:
   • Kafka의 Log Compaction은 백그라운드에서 실행되는 프로세스입니다. log.cleaner라는 백그라운드 쓰레드가 로그 파일을 모니터링하고, 설정된 조건에 따라 컴팩션을 수행합니다.
2. 조건 충족 시 실행:
   • 앞서 언급한 조건들, 예를 들어 log.cleaner.min.cleanable.ratio 이상의 데이터가 정리 가능할 때, delete.retention.ms가 지나 오래된 메시지가 보존되지 않아도 될 때, 또는 디스크 공간이 부족해질 때 등의 상황에서 컴팩션이 발생합니다.
   • 이 조건들은 Kafka 클러스터의 설정에 따라 달라질 수 있으며, 이를 통해 컴팩션 빈도와 강도를 조절할 수 있습니다.
3. 타이밍 제어:
   • min.compaction.lag.ms와 max.compaction.lag.ms와 같은 설정을 통해 컴팩션이 발생할 수 있는 최소 및 최대 지연 시간을 제어합니다. 이를 통해 메시지가 너무 일찍 삭제되지 않도록 하고, 너무 오래 유지되지 않도록 합니다.

요약

Log Compaction이 트리거되는 정확한 시점은 위의 설정값과 시스템의 현재 상태에 따라 달라집니다. 특정 파티션에서의 로그 세그먼트가 충분히 클 때, 정리 가능 비율이 높을 때, 디스크 사용량이 제한에 도달했을 때 등 여러 조건이 컴팩션을 유도합니다. 이러한 조건을 충족할 때마다 Kafka는 자동으로 컴팩션을 수행하여 로그 데이터를 최적화하고 저장 공간을 회수합니다

 

Log Compaction의 단계별 과정

1. 로그 세그먼트 선택
   • Log Compaction 대상 선정: log.cleaner.min.cleanable.ratio 등의 설정을 기반으로, 로그 세그먼트가 얼마나 정리 가능한지 확인합니다. 정리 가능 비율이 기준치 이상인 세그먼트가 컴팩션 대상이 됩니다.
   • 순차적 선택: 컴팩션은 Kafka 브로커에 있는 여러 파티션의 세그먼트들 중에서 순차적으로 선택되어 수행됩니다.
2. 로그 세그먼트 읽기
   • 세그먼트 로드: 대상 세그먼트가 선택되면, 컴팩션 프로세스는 해당 세그먼트를 메모리로 로드합니다.
   • 레코드 스캔: 세그먼트에 있는 모든 레코드를 키(key) 순서로 스캔합니다. 이 과정에서 동일한 키를 가진 레코드 중 가장 최근의 레코드를 식별합니다.
3. 레코드 필터링
   • 최신 레코드 식별: 각 키에 대해 가장 최신의 레코드만 유지하고, 오래된 레코드를 제거할 대상으로 표시합니다.
   • 삭제된 레코드 처리: 레코드가 삭제된 경우, tombstone이라고 불리는 삭제 마커가 남습니다. 이 마커는 delete.retention.ms 동안 유지된 후에야 완전히 제거됩니다.
4. 새로운 세그먼트 작성
   • 새로운 컴팩션 세그먼트 생성: 유지할 최신 레코드들로 새로운 로그 세그먼트를 작성합니다.
   • 이전 세그먼트 교체: 새로운 세그먼트가 생성되고, 기존 세그먼트는 더 이상 필요하지 않은 레코드를 제거하기 위해 교체됩니다.
5. 세그먼트 파일 교체
   • 원자적 교체: 새로운 세그먼트 파일을 기존 파일과 교체하는 과정은 원자적으로 수행됩니다. 이는 교체 중 오류가 발생해도 데이터 손실이나 불일치가 발생하지 않도록 보장합니다.
6. 불필요한 데이터 제거
   • 디스크 공간 회수: 컴팩션 후 남은 불필요한 세그먼트 파일은 삭제되어 디스크 공간이 회수됩니다. 이는 delete.retention.ms 설정에 따라 삭제 마커도 제거하는 단계가 포함됩니다.
7. 레코드 재사용
   • 레코드 재사용: 일부 오래된 레코드가 유지되는 경우, 이들은 새로운 세그먼트 파일에 포함됩니다. 이는 해당 레코드가 유효 기간 내에 있고 여전히 최신 정보이기 때문입니다.

Low-level 상세 설명

• 키 정렬 및 해시 테이블: Kafka는 로그 세그먼트 내의 레코드를 정렬하여 저장하고, 각 키에 대한 최신 레코드를 추적하기 위해 해시 테이블을 사용합니다. 이는 특정 키에 대한 최신 레코드를 빠르게 찾을 수 있게 합니다.
• tombstone 레코드: 삭제된 레코드의 표시로, 삭제된 데이터가 실제로 제거되기 전에 해당 레코드가 삭제되었음을 나타내는 마커 역할을 합니다.
• 원자적 교체: 새로운 세그먼트가 준비되면, 기존 파일을 삭제하고 새 파일로 교체하는 과정은 중간 단계에서의 실패가 발생하더라도 데이터 손실을 방지하기 위해 원자적으로 이루어집니다.

Log Compaction은 이러한 단계를 통해 Kafka 클러스터 내에서 저장 공간을 최적화하고, 시스템 성능을 유지하며, 데이터의 일관성을 보장합니다.