Kafka Log Compaction 알아보기

- 목차

 

들어가며.

카프카 토픽은 데이터의 용량과 시간 조건을 기준으로 데이터를 삭제합니다.

이를 Retention 이라고 하는데요.

retention.ms, retention.bytes 와 같은 설정을 통해서 카프카는 토픽의 데이터의 저장 기간을 지정할 수 있습니다.

카프카 토픽의 기존 설정으로 retention.ms 는 7일 그리고 retention.bytes 는 제한이 없습니다.

즉, 기본 설정의 토픽을 생성하게 되면 7일 주기로 토픽에 추가된 데이터는 삭제되고 용량과 관련된 제약은 없다고 생각하시면 됩니다.

 

아래 이미지처럼 retention.ms 가 7일인 상황에서 7일이 경과된 Log Sgement 파일들은 하나씩 삭제가 됩니다.

00000000.log, 00010000.log, 00033000.log 세개의 로그 파일이 존재하고,

00000000.log, 00010000.log 로그파일이 7일이 경과된 파일인 경우에

00033000.log 파일 이외의 7일이 경과된 로그 파일들은 Retention 의 적용 대상이 됩니다.

 

 

 

하지만 위와 같이 Retention 의 적용은 Topic 의 Cleanup Policy 정책에 따릅니다.

Cleanup Policy 의 기본 정책은 Delete 이며, Cleanup Policy 가 Delete 이기 때문에 Retention 에 의해서 로그 파일들은 삭제될 수 있습니다.

반면, 오늘 알아볼 내용인 Compact Cleanup Policy 가 존재합니다.

이는 동일한 Key 를 가지는 Record 에 대해서 최신의 Record 만을 유지시키는 Cleanup Policy 입니다.

 

이어지는 글에서 Topic 의 Compaction 에 대해서 자세히 알아보도록 하겠습니다.

 

Log Segment 란 ?

https://westlife0615.tistory.com/514

Kafka Log Segment 알아보기

 

카프카 토픽은 메시지를 저장하는 주체입니다.

토픽은 내부적으로 파티션이라는 단위로 나뉘게 되는데, 파티션 또한 로그 파일이라는 단위로 저장됩니다.

아래 이미지가 Topic - Partition - Log Segment 를 표현하는 이미지입니다.

Log Segment 는 실질적인 레코드를 보관하는 로그 파일이라고 생각하시면 됩니다.

레코드의 양이 많아질수록 Log Segment 의 갯수가 증가합니다.

 

 

Rolling Update.

카프카의 로그 파일은 시간과 용량 조건에 따라서 생성됩니다.

이는 Retention 과 유사합니다.

Retention 또한 retention.ms 와 retention.bytes 에 설정된 시간과 용량 조건에 따라서 데이터의 저장 기간이 결정되죠.

Log Segment 도 segment.ms 와 segment.bytes 라는 시간과 용량 조건이 존재합니다.

segment.ms 에 설정된 시간이 지나면 새로운 Log Segment 파일이 생성됩니다.

그리고 segment.bytes 에 설정된 용량을 초과하게 되면 새로운 Log Segment 파일이 생성되죠.

 

아래의 명령어는 test-segment 라는 토픽을 생성하는 명령어입니다.

그리고 segment.bytes 를 1KB, segment.ms 는 30초로 설정하였습니다.

test-segment 토픽을 생성하여 Log Segment 의 생성 패턴을 파악해보겠습니다.

 

kafka-topics --bootstrap-server localhost:9092 --create --topic test-segment \
--replication-factor 1 --partitions 1 --config cleanup.policy=delete \
--config segment.bytes=$((1 * 1024)) --config segment.ms=$((30 * 1000))

 

아래의 사진은 생성된 test-segment 의 이미지입니다.

segment.bytes 를 1KB, segment.ms 는 30초로 설정하였고, Partition & Replication Factor 는 각각 1 로 설정하였습니다.

 

 

그리고 아래와 같이 test-segment 토픽의 0 번 파티션은 여러개의 .log 파일들을 가지게 됩니다.

XXX.log 와 같은 파일명을 가지게 되구요.

XXX 에 해당하는 파일명은 log 파일의 첫번째 Record 의 Offset 을 의미합니다.

각 log 파일은 segment.bytes 값인 최대 1KB 입니다.

 

 

segment.ms 기준으로 Log Segment 생성하기.

test-segment 토픽은 segment.ms 를 30초로 설정하였습니다.

그래서 30초마다 새로운 Record 가 유입된다면, 새로운 Log Segment 를 생성합니다.

 

아래와 같은 Pseudo Code 로 60초 주기로 하나의 레코드를 test-segment 토픽으로 생성하게되면,

segment.ms 인 30초를 주기로 하나의 Record 마다 하나의 Log Segment 가 생성될 것입니다.

for (long i = 1; i <= 1000000; i++) {
  Thread.sleep(60 * 1000);
  producer.send(new ProducerRecord<>(topic,"1", i + ""));
  producer.flush();
}

 

그리고 생성된 Log Segment 를 확인해보면 아래 이미지와 같이 70byte 크기의 Log Segment 파일이 계속 생성됨을 알 수 있습니다.

.log 파일의 파일명이 저장된 첫번째 레코드의 Offset 을 의미하는데요.

Offset 이 1839, 1840, 1841 인 레코드가 각각 하나의 Log Segment 파일에 저장됨을 알 수 있습니다.

 

 

 

 

Log Compaction 이란 ?

지금까지 Retention 과 Log Segment 에 대해서 알아보았습니다.

Log Compaction 에 대해서 알기 위해선 위 두 개념이 선행되어야합니다.

Log Compaction 은 동일한 Key 를 가진 레코드들이 있다면 가장 최신의 Record 만을 남기고

나머지 Old Record 들을 제거합니다.

여기서 중요한 포인트는 동일한 Key 와 Old Record 의 제거입니다.

 

먼저 Log Compaction 이 어떻게 동작하는지 먼저 확인해보겠습니다.

아래의 명령어는 cleanup.policy 를 compact 로 설정한 토픽을 생성하는 명령어입니다.

 

kafka-topics --bootstrap-server localhost:9092 --create --topic compacted \
--replication-factor 3 --partitions 3 --config cleanup.policy=compact \
--config segment.bytes=$((1 * 1024)) --config segment.ms=$((30 * 1000))

 

아래와 같이 compacted 토픽이 생성되었습니다.

segment.bytes 는 1KB 로 설정되었구요. Producer 가 생성한 레코드를 1KB 만큼 하나의 로그파일에 저장할 수 있습니다.

그리고 segment.ms 는 30 초로 설정하였습니다. 30초 주기로 새로운 로그파일이 생성됩니다.

 

 

아래는 Kafka Producer 코드 예시입니다.

segment.ms 가 30초이기 때문에 1분 주기로 데이터를 생성하여, 하나의 로그파일에 하나의 레코드가 저장되도록 구성하였습니다.

그리고 모든 레코드의 key 는 "1" 로 설정하여 Compaction 이 발생하도록 의도하였습니다.

 

  public static void main (String[] args) throws InterruptedException {
    Properties properties = new Properties();
    properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:29091,localhost:29092,localhost:29093");
    properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
    properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
    String topic = "compacted";
    KafkaProducer<String, String> producer = new KafkaProducer<>(properties);
    for (long i = 1; i <= 1000000; i++) {
      Thread.sleep(60 * 1000);
      producer.send(new ProducerRecord<>(topic,"1", i + ""));
      producer.flush();
    }
    producer.close();
  }

 

아래의 이미지들처럼 segment.bytes, segment.ms 를 기준으로 Compaction 과정이 주기적으로 진행됩니다.

 

 

 

Log Compaction 과 Log Segment 의 관계.

Log Compaction 의 대상은 현재 Rolling Update 중인 Log Segment 가 아닌 과거의 Log Segment 들입니다.

그래서 segment.ms, segment.bytes 설정에 의해서 새로운 Log Segment 가 생성될 때에 Log Compaction 이 백그라운드에서 동작합니다.

아래의 이미지처럼 Partition 1 - Log Segment 3 이라는 새로운 Log Segment 가 생성될 때에

기존의 Partition 1 - Log Segment 1, Partition 1 - Log Segment 2 는 Log Compaction 의 대상이 됩니다.

그래서 새로운 Log Segment 3 으로 Compaction 되어 Merge 됩니다.

 

 

 

 

마치며.

 

Compact Cleanup Policy 는 대표적으로 __consumer_offset__ 토픽이 이러한 정책을 따릅니다.

__consumer_offset__ 는 각 Consumer Group 의 Offset 이 저장되는 토픽으로써 최신의 데이터만이 의미가 있습니다.

이러한 느낌으로 Latest Event 의 역할이 중요한 시스템에서 활용하기에 알맞은 Cleanup Policy 라고 생각합니다.

 감사합니다.