[Kafka] Topic, Partition, Segment, Segment 관리, Offset

카프카(kafka)에서 다뤄지는 주요 개념들인 토픽(Topic), 파티션(partition), 세그먼트(Segment), 오프셋(Offset)에대해서 그리고 세그먼트(segment)의 경우 관리하는 법 까지 알아보도록 하겠습니다.

 

토픽(Topic)

---

Topic

Event는 무엇인가 발생한 사실에 대한 record이며 record 또는 message라고 공식 document에서는 말합니다.

Events are organized and durably stored in topics.
Very simplified, a topic is similar to a folder in a filesystem, and the events are the files in that folder.

이 Event가 구조화되어 topic에 영구적으로 저장이 됩니다.

매우 간단하게 Event, Message는 file이며 Topic은 folder입니다.

즉 Topic은 Consumer가 가져가는 Event, Message를 보관하고 있는 folder라고 생각하시면 됩니다.

 

파티션(Partition)

---

분산처리 시스템으로 잘 알려진 kafka에서 생성한 topic이 클러스터 내에서 하나의 machine, 또는 하나의 node에서만 사용된다면??

topic이 배포할 수 있는 cluster내에 작동하는 node의 개수보다 많아질 수 없기 때문에 Kafka를 스케일링하는 것에 대한 근본적인 한계로 작용할 수 있습니다.

병렬 처리가 가능하며 스케일링하여 대규모 메시지 처리를 할 수 있도록 토픽 하나를 여러 개로 나눈 것이 파티션이라고 합니다.

하나를 여러 개로 나누면 분산 처리도 가능합니다.

나눈 파티션 수 만큼 컨슈머를 연결할 수 있습니다.

topic과 partition의 관계

파티션 번호는 0번부터 시작합니다.

파티션은 초기 생성후 그 이상으로 늘릴 수는 있지만 반대로 한번 늘린 파티션은 줄일 수 없습니다.

 

파티션수를 2, 4로 작게 생성한 뒤 컨슈머에 LAG 등을 모니터링하면서 조금씩 늘리는 게 좋습니다.

LAG : 프로듀서가 보낸 메시지수 - 컨슈머가 가져간 메시지수

즉 LAG라는 지표를 통해 컨슈머에 지연이 있는지 없는지를 확인 가능합니다.

 

세그먼트(Segment)

---

Producer에 의해서 브로커로 전송된 메시지는 토픽의 파티션에 저장되며, 각 메시지들은 세그먼트라는 로그 파일의 형태로 브로커의 로컬 디스크에 저장됩니다.

출처 :  https://docs.cloudera.com/runtime/7.2.10/kafka-overview/topics/kafka-overview-logs-and-log-segments.html

각 partition 마다 n개의 segment들이 존재합니다.

 

kafka 브로커, topic, partition, segment 구조

1. Producer는 kafka의 A라는 Topic으로 메시지를 전달합니다.
2. A Topic은 partition이 하나여서 0이며 Producer로부터 받은 메시지를 Partition의 Segment log파일에 저장합니다
3. Broker의 Segment log file에 저장된 메시지는 Consumer가 읽어 갈 수 있습니다.
4. Consumer는 A Topic을 컨슘 해서 해당 토픽 내 파티션 0의 segment log file에서 메시지를 가져옵니다.

 

 

kafka container에서 segment를 확인해 보도록 하겠습니다.

/opt.kafka/bin $ kafka-topics.sh --create --topic log-message-test 
--bootstrap-server kafka1:9092 --replication-factor 3 --partitions 2

partition값이 2인 topic을 생성합니다.

 

producer를 실행시킨 뒤 메시지를 두 가지를 보냅니다.

• test message
• test message number2

 

cd /kafka/kafka-logs-${hashvalue}
ls

 

경로 이동을 하고 ls 명령어로 확인하면 topic 폴더가 존재합니다.

topic명 - 파티션 숫자 현재의 경우 토픽생성 시 partion 값을 2로 주었기 때문에 두 가지가 생성되어 있습니다. (0부터 시작합니다.)

ex) log-message-test-0 , log-message-test-1

 

log-message-test-0으로 들어가서 확인해 보면.

cd log-message-test-0
ls

 

4가지가 존재합니다.

00000000000000000000.index : 로그 세그먼트에 저장된 위치와 offset 정보를 기록합니다.

00000000000000000000.log : 실제 메시지들이 저장되는 파일입니다.

00000000000000000000.timeindex : 메시지의 타임스탬프를 기록하는 파일입니다.

leader-epoch-checkpoint : kafka에서 컨트롤러가 이전에 할당한 리더의 수를 나타냅니다.

• 새 리더를 선택하고 leader-epoch를 1씩 증가시키며 모든 replicas와 공유합니다.

 

xxd 00000000000000000000.log

 

hex dump를 보여주는 xxd 명령어를 통해 00000000000000000000.log 파일의 내용을 살펴보겠습니다.

 

test message number2를 확인할 수 있습니다.

 

log-message-test-1으로 들어가서 hex dump 본다면

test message를 확인할 수 있습니다.

 

로그 세그먼트 관리 방법

---

로그 세그먼트에는 메시지의 내용만 저장되는 것이 아니라 메시지의 key, value, offset, message size와 같은 정보가 함께 저장됩니다.

하나의 로그 세그먼트 크기가 너무 커지면 파일 관리가 어렵기 때문에 최대 크기는 1GB 기본값으로 설정되어 있습니다.

 

1GB보다 크기가 커진다면?

Rolling 전략을 사용
ex) 하나의 로그 세그먼트에 계속 메시지를 덧붙이다가 1GB에 도달하면 close한뒤 새로운 로그 세그먼트를 생성하는 방식

 

1GB 크기의 로그 세그먼트 파일이 무한히 늘어날 경우를 대비해 관리 계획을 수립해야 합니다.

관리 계획은 크게 로그 세그먼트 삭제와 컴팩션으로 구분할 수 있습니다.

 

로그 세그먼트 삭제

---

broker 설정 파일인 server.properties에서 log.cleaanup.policy 가 delete로 명시되어야 합니다.

 

위의 log-message-test 예시를 이어서 활용하며 retention.ms 옵션을 조정해서 메시지를 삭제해 보겠습니다.

 

opt/kafka/bin $ kafka-configs.sh --bootstrap-server kafka1:9092 —-topic log-message-test 
--add-config retentton.ms=0 --alter

describe 옵션을 통해서 retention 옵션이 추가된 것을 확인할 수 있습니다.

retention.ms = 0 이란 로그 세그먼트 보관 시간이 해당 숫자보다 크면 segment를 삭제한다는 명령입니다.

기본값은 5분 주기 이므로 5분 간격으로 세그먼트 파일을 체크하면서 삭제 작업을 수행합니다. 즉시 삭제되는 것이 아니라 약 5분 후에 삭제 작업이 일어납니다.

 

이전

 

이후

topic partition이 2이기 때문에 log-message-test 0, log-message-test1 둘 다 0000000000000000000로 시작하는 파일은 모두 삭제되고 00000000000000000001 로 시작하는 파일이 새로 생성된 것을 볼 수 있습니다.

신규 로그 세그먼트 파일은 위와 같이 기존 로그 세그먼트 파일에서 1이 증가한 파일명으로 생성됩니다

1은 offset을 의미합니다.

즉 카프카는 로그 세그먼트 파일을 생성할 때 offset 시작 번호를 활용해서 파일 이름을 생성하는 규칙을 따릅니다.

 

retention.ms 옵션 제거

opt/kafka/bin $ kafka-conftgs.sh --bootstrap-server kafkal:9092 —-topic log-message-test 
--delete-config retention.ms —alter

describe 명령어로 retention값이 없어진 것을 확인할 수 있습니다.

카프카의 관리자는 topic 마다 보관 주기를 조정해서 얼마만큼의 기간 동안 카프카에 로그를 저장할지 결정하고 관리할 수 있습니다.

retention.ms 옵션을 설정하지 않으면 카프카의 server.properties에 적용된 옵션값이 자동 적용되며 설정값은 7일입니다.

 

로그 컴팩션

---

로그를 삭제하지 않고 컴팩션하여 보관할 수 있습니다.

카프카에서 로그 세그먼트를 컴팩션하면 메시지의 키값을 기준으로 마지막 데이터만 보관하게 됩니다.

 

대표적으로 __consumer_offset 토픽입니다.

카프카 내부 토픽으로 컨슈머 그롭 정보를 저장하는 토픽입니다.

각 컨슈머 그룹에 해당 컨슈머가 어디까지 읽었는지를 나타내는 offset 커밋 정보인데 key(컨슈머 그룹명, 토픽명)-value(오프셋 커밋 정보) 형태로 메시지가 저장됩니다.

 

__consumer_offset 예시

ex) CG01 Consumer Group 이 T01 Topic을 Consume 하고 첫 번째 메시지를 커밋한다면

• key : CG01(컨슈머 그룹), T01(토픽명)
• value : 1(offset)

key, value가 __consumer_offset 토픽에 저장됩니다.

 

ex) 시간이 지나 CG01 Consumer Group 이 T01 Topic을 Consume 하고 2번째 메시지를 읽고 커밋한다면
value값이 2로 변경되며, 이후 3번째 메시지를 읽고 커밋한다면 value 값이 3으로 변경됩니다.

 

이때 로그 컴팩션이 동작한다면 마지막에 읽은 키값이 (CG01, T01)인 메시지의 마지막 데이터인 3만 남게 됩니다.

컨슈머에 입장에서는 항상 마지막 커밋된 offset 정보가 중요하기 때문에 과거 커밋 정보는 삭제돼도 무방합니다.

 

로그컴팩션은 위와 같이 과거 정보는 중요하지 않고 가장 마지막 값이 필요한 경우에 사용합니다.

producer가 카프카로 메시지를 전송할 때 메시지에는 메시지의 key와 value를 같이 전송하게 됩니다.

value는 필수 값이지만 key는 필수 값이 아닙니다 따라서 로그 컴팩션 기능을 사용하고자 한다면 카프카로 메시지를 전송할 때 필수값으로 전송해야 합니다.

 

출처 : https://kafka.apache.org/documentation.html#compaction

로그 컴팩션 예시입니다 컴팩션 후 모두 마지막 메시지만 저장됩니다.

그렇다면 로그 컴팩션의 장점은 무엇일까요?

장점은 바로 빠른 장애 복구입니다.

전체 로그를 복구하지 않고 메시지의 키를 기준으로 최신의 상태만 복구합니다.

전체 로그를 복구할 때보다 복구 시간을 줄일 수 있다는 장점이 있습니다.

하지만 로그 컴팩션이 실행되는 동안 브로커의 과도한 입출력 부하가 발생할 수 있으니 유의해야 하며 반드시 브로커의 리소스 모니터링도 병행해야 합니다.

 

로그 컴팩션 관련 옵션

옵션 이름	옵션값	적용 범위	설명
cleanup.policy	compact	토픽의 옵션으로 적용	토픽 레벨에서 로그 컴팩션을 설정 할 때 적용하는 옵션입니다.
log.cleanup.policy	compact	브로커의 설정 파 일에 적용	브로커 레벨에서 로그 컴팩션을 설 정할 때 적용하는 옵션입니다.
log.cleaner.min. compaction.lag.ms	0	브로커의 설정 파 일에 적용	메시지가 기록된 후 컴팩션하기 전 경과되어야 할 최소 시간을 지정합 니다. 만약 이 옵션을 설정하지 않 으면, 마지막 세그먼트를 제외하고 모든 세그먼트를 컴팩션할 수 있습 니다.
log.cleaner.max. compaction.lag.ms	9223372036854775807	브로커의 설정 파 일에 적용	메시지가 기록된 후 컴팩션하기 전 경과되어야 할 최대 시간을 지정합 니다.
log.cleaner.min. cleanable.ratio	0.5	브로커의 설정 파 일에 적용	로그에서 압축이 되지 않은 부분을 더티(dirty)라고 표현합니다. ‘전체 로그’ 대비 ‘더티’의 비율이 50%가 넘으면 로그 컴팩션이 실행됩니다.

 

오프셋(Offset)

---

Offset

파티션에 메시지가 저장되는 위치를 Offset이라고 합니다.

Offset은 순차적으로 증가하는 숫자 형태로 되어 있습니다.

출처 : https://www.geeksforgeeks.org/topics-partitions-and-offsets-in-apache-kafka/

각 partition 마다 순차적으로 증가하는 숫자들을 볼 수 있는데 이 숫자가 offset입니다.

• offset은 각 partition 마다 고유한 숫자입니다.
• offset을 통해서 메시지 순서를 보장하고 이 offset을 통해서 consumer에서는 마지막으로 읽은 위치를 알 수 도 있습니다.

 

참조

---

Apache Kafka

Kafka: Is it a Topic or a Queue?

what is a role of leader epoch checkpoint in log segment of Kafka partition?

실전 카프카 개발부터 운영까지