카프카 살펴보기(1부)

본 글에서는 메시지의 저장과 소비만을 다룬다.
- keyword: topic, partition, offset, partitioner, segment, clean-up policy

들어가며

이벤트는 키-밸류 구조로 되어있는, 무슨 일이 일어났는지에 대한 사건의 기록이다.

사건을 기록하고 꺼내어봄으로써 각 어플리케이션의 요구에 맞게 특정한 행동을 수행할 수 있다.

키는 기록에 대한 구분자로, 어떤 user 가 될 수도 있고, 번호가 될 수도 있다. 하나의 키에 여러 가지 사건이 발생될 수 있으므로 유니크할 필요도 없고 반드시 있어야만 하는 것도 아니다. 이 키에 대한 내용을 밸류로 갖고, 이벤트 메시지는 사실 키-밸류와 함께 타임스탬프, 메타정보인 헤더를 갖는다. 이벤트는 하나의 기록이므로 당연히 변경되지 않는다.

카프카는 실시간 데이터를 처리한다. 우리가 일기를 쓰듯이 하루에 한 번 지나간 사건에 대해 기록하는 것이 아니다. 상시 특정 대상에 대한 변화 지점을 계속해서 포착하는 것이다. 카프카는 이러한 실시간 이벤트를 수집하고, 그것들을 받아서 바로 처리할 수 있게끔, 또는 나중에 찾아볼 수 있게끔 잘 구성해서 저장한다. 각 이벤트들은 기록 자체로는 서로 상관 없는 개별적인 사건이 될 뻔했지만, 이를 잘 구성함으로써, 데이터의 지속적인 흐름과 해석을 보장해주고 있다. 실시간 정보를 주고 받는 상시 접속 시대에, 자동화된 메시징 처리를 가능하게 하는 이벤트 스트리밍이 카프카의 기본 컨셉이다.

그럼 어떻게 잘 구성해서 저장하길래 다양한 애플리케이션이 필요에 따라 효과적으로 데이터를 찾고 처리할 수 있게 하는가?

 

데이터의 저장

메시지 저장 순서 보장

이벤트는 토픽에 저장된다. 토픽은 말 그대로 특정 주제인데, 그 주제에 해당하는 메시지들을 저장한다. 카프카는 메시지를 저장, 관리, 분산 처리를 하는 카프카 서버(브로커)와 이 메시지를 발행하고 수신하는 클라이언트인 프로듀서, 컨수머로 구성된다. 프로듀서가 발행한 메시지는 브로커에 쌓인다.

실제로 데이터가 저장되는 곳은 파티션이다.

 

 

 

 

https://curiousjinan.tistory.com/entry/understand-kafka-partitions

 

하나의 토픽은 여러 개의 파티션을 가질 수 있다. 각 파티션에는 특정 컨수머가 동적으로 할당되는데, 파티션이라는 개념이 없었다면, 여러 개의 메시지 발행자, 소비자가 메시지 읽기/쓰기를 하기 위해 계속 줄서있어야 했을 것이다. 파티션은 특정 주제 내에서의 분산을 가능하게 한다.

그렇다면 메시지 순서는 어떻게 보장되지? 이걸 이해하기 위해 필요한 개념은 오프셋이다. 오프셋은 파티션에 저장되는 메시지의 위치이다.

오프셋은 파티션 내에서 고유하다. 즉, 파티션이 달라지면 오프셋은 달라진다. 각 메시지는 토픽과 파티션, 오프셋을 가짐으로써 저장 순서가 보장된다.

 

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

 

다시 말하면, 파티션 밖에서의 오프셋은 의미가 없다. 위 그림에서 7번 메시지는 파티션 0, 1, 2 에서 모두 다른 메시지이다. 메시지를 저장된 순서에 따라 처리하려면, 같은 파티션으로 배치해야 한다.

 

파티셔닝

카프카는 이를 위해 파티셔닝이라는 개념을 가지고 있다. 어떤 파티션에 배치할 것인가를 결정하는 전략인데, 총 세 가지의 방식이 있다.

첫 번째로, 키 기반 파티셔닝은 메시지가 키를 가지고 있을 때, 같은 키는 같은 파티션에 할당하는 전략이다.

두 번째로, 스티키 파티셔닝은 배치 사이즈 또는 시간으로 결정된다. 배치 설정에 따라 크기가 꽉 차면 / 시간이 되면 그 묶음은 한 파티션에 할당한다. 만약 배치가 차지 않았다면, 이전 레코드와 동일한 파티션에 발행되고, 새로 생성된 배치는 무작위로 할당된다.

세 번째는 라운드 로빈 방식이다. 카프카 2.4 이전까지는 키가 없을 때의 기본 전략이었다.

위 동작을 수행하는 것이 파티셔너이다.

파티셔너는 프로듀서의 영역이다. 메시지가 발행될 때 어떤 파티션에 배치될지가 결정된다. 필요하다면 개별 어플리케이션에서 Partitioner 를 구현하여 커스텀 파티셔너를 만들어도 된다.

키 기반 파티셔닝을 데이터 스트림 관점에서 보자면, 파티션은 주제별로 묶인 이벤트를 또 한 번 묶은 최후의 데이터 단위인 셈이다.

https://www.youtube.com/watch?v=kj9JH3ZdsBQ&t=22s

 

파티션의 구성

앞서 언급했듯, 예상되는 부하에 따라 토픽 별로 파티션 개수를 정할 수 있다.

몇 개가 적절할까? 우선 파티션은 늘리는 건 가능해도, 줄일 수는 없다. 그렇다면 최초에 넉넉하게 만들되, 지나치게 많지 않게 해야 한다. 또, 키 기반 파티셔닝을 사용할 때, 파티션의 개수를 늘리면 파티셔닝 계산이 깨진다. 순서 보장이 안 될 수 있다. 때문에 늘리는 것도 신중하게 해야 한다.

어차피 파티션 수가 더 적으면 특정 컨수머는 유휴 상태에 들어간다. 반대로 지나치게 많으면 하나의 컨수머가 처리해야 하는 파티션 수가 너무 늘어나게 된다. 쓰기 속도, 파티션 키 분산도 신경 써야 한다.

결국 잘, 해야 한다는 건데, https://docs.cloudera.com/runtime/7.2.18/kafka-performance-tuning/topics/kafka-tune-sizing-partition-number.html 에 따르면 파티션 수 = max(목표 처리량 / 프로듀서 처리량, 목표 처리량 / 컨슈머 처리량) 이런 공식을 제공하고 있긴 하다.

메시지의 삭제

메시지는 최대 저장 한도가 있다. 기본적으로 7일, 무제한이 최대 한도이다. 하지만 서버에 메시지가 쌓이기만 한다면 문제가 생길 것이다.

어떤 메시지를 어떻게 삭제할 것인가? 카프카의 메시지의 삭제 정책은 두 가지가 있다.

• cleanup.policy=delete(default): 정해진 용량이나 시간을 초과하면 삭제
  • retention.ms / log.retention.ms
  • retention.bytes / log.retention.bytes
• cleanup.policy=compact: key 기준 최신 메시지만 남기고 삭제

https://kafka.apache.org/documentation

• cleanup.policy=delete,compact: delete 와 compact 를 조합

다만 메시지 삭제는 개별 메시지 단위가 아닌 ‘세그먼트’ 단위로 이루어진다. 따라서 실제 메시지 보존 기간은 설정값보다 길어질 수 있음에 유의해야 한다.

자세한 얘기를 하기 전에, 세그먼트가 뭔지부터 보자.

 

세그먼트

세그먼트는 메시지가 저장되는 묶음이다. 아까 메시지는 파티션에 저장되고, 파티션 내에서만 순서가 유지된다고 했는데, 이건 또 무슨 소리일까?

메시지가 파티션에 저장된다고 하면, 파티션에 차곡차곡 순서대로 쌓을 것 같지만, 사실은 세그먼트 단위로 만들어진 파일을 차곡차곡 쌓게 된다. 아래의 그림과 같이, 설정된 크기 또는 시간(segment.ms / segment.bytes)에 따라 세그먼트가 생성된다.

https://rohithsankepally.github.io/Kafka-Storage-Internals/

 

토픽 내 파티션에 메시지를 발행한다면, 실제 세그먼트는 아래와 같이 생긴다.

메시지는 .log 라는 하나의 파일에 쌓인다. (*.index, *.timeindex 파일도 생긴다)

파일명은 시작 offset 을 의미한다.

00000000000000000000.* - Capture storage of messages with 0 <= offset <35.
00000000000000000035.* - Capture storage of messages with offset >=35

자세한 내용은 https://rohithsankepally.github.io/Kafka-Storage-Internals/ 이 글을 보라.

삭제의 대상

log 파일을 통해 우리는 delete 를 하든, compact 를 하든 삭제가 세그먼트 단위로 이뤄지는 이유에 대해 알 수 있다.

데이터는 로그의 헤드 부분에 쌓인다. 자연히 오래된 메시지는 테일 부분에 오래된 순으로 쌓여 있다.

세그먼트는 크기 또는 시간에 따라 구분된다고 했다. 이미 크기 또는 시간이 꽉 찬 테일 부분은, 여러 개의 세그먼트로 나뉘어져 있을 것이다. 반면 헤드의 세그먼트에는 계속해서 새로운 데이터가 쌓이게 된다.

https://docs.confluent.io/kafka/design/log_compaction.html

 

아래의 그림이 있을 때, 삭제 후보는 활성화된 세그먼트를 제외하고, 아래의 1, 2, N-1 이다.

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

 

실제 삭제 대상은 delete 이냐, compact 이냐에 따라 다르다.

 

 

delete 정책은

• 보존 기간이나 크기를 초과한 닫힌 세그먼트를 대상으로 한다.
• 오래된 세그먼트부터 순차적으로 삭제한다.
• 설정된 보존 기준을 초과하지 않는 세그먼트는 그대로 유지된다.
• 위의 그림에서, Segment N 이 활성화된 세그먼트이고, 보존 기준을 초과한 세그먼트가 Segement 1 뿐이라면, 2, N-1, N 은 그대로 유지된다.

세그먼트가 생기는 시간 / 사이즈가 있고, 보존 기간 / 사이즈가 있기 때문에 어떤 메시지가 삭제되느냐는 예상과는 조금 다를 수 있다.

아래의 그림을 보면, 세그먼트 사이즈가 보존 기간보다 크고, 체크 스레드가 보존 기간을 초과한 세그먼트를 삭제하고 있다.

https://www.redpanda.com/guides/kafka-performance-kafka-logs

 

 

compact 정책은

• 닫힌 세그먼트를 대상으로 한다.
• 각 키에 대해 가장 최신 값만을 유지하고, 이전 버전의 데이터는 제거한다.
• 압축 과정에서 새로운 세그먼트가 생성되며, 이 세그먼트에는 각 키의 최신 값만 포함된다.

compact 방식일 때의 대상은 보존 기간의 초과가 아니라, 아래 세 가지 설정에 의해 결정된다.

• log.cleaner.delete.retention.ms controls how long the records are retained after they are confirmed for deletion.
• log.cleaner.min.compaction.lag.ms defines the minimum age of the records that are subject to compaction.
• log.cleaner.max.compaction.lg.ms defines the maximum age of records that can remain in the system without compaction.

https://developer.confluent.io/courses/architecture/compaction/

 

 

만약 compact 방식인데 키가 null 이면 어떻게 될까? 키가 null 인 레코드는 전송되지 않는다(https://blog.voidmainvoid.net/505 ).

더 자세하게 알고 싶으면 https://developer.confluent.io/courses/architecture/compaction/ 여기에 아주 잘 설명돼 있다.

 

1부를 마치며

 

카프카는 로그 기반 구조를 취함으로써, 한 번 쓰여진 데이터는 변경이 불가능하게 한다. 또한 테일 영역의 데이터만 삭제함으로써 삭제를 용이하게 하며, 끝에만 저장하니까 속도도 빠르다(https://docs.confluent.io/kafka/design/file-system-constant-time.html ).

카프카는 이러한 방식으로 데이터를 저장함으로써, 이벤트 스트리밍이라는 컨셉을 구현했다. 또, 이 글에서 다루지는 않았지만, 파티션의 복제를 통해 분산 처리를 지원하는데, 로그 기반의 순차적인 시스템이 이것을 가능하게 한다.

여기까지, 실시간으로 변화하는 현재의 사건들을 여러 클라이언트가 대응할 수 있도록 한 카프카의 컨셉이 어떻게 실현되었는지 살펴봤다.

 

메시지 소비

to be continued