[Kafka] 기본 개념 및 이해

Apache Kafka란?

이벤트 스트리밍 플랫폼

실시간으로 흐르는 이벤트 스트림을 받아주고 이벤트를 필요로 하는 곳으로 이벤트를 전달한다.

이벤트? 비즈니스에서 일어나는 모든 일(데이터)을 의미

ex) 웹사이트에서 뭔가 클릭하는것. 돈을 송금하는 것 등

 

이벤트는 bigdata의 특징을 가짐 -> 비즈니스의 모든 영역에서 광범위하게 발생하기 때문

이벤트 스트림은 연속적인 많은 이벤트들의 흐름을 의미

 

이렇게 많은 이벤트들을 처리하기 위해 LinkedIn에서 개발된 것이 Apache Kafka

 

3가지 특징

1. 이벤트 스트림을 안전하게 전송 (publish & Subscribe 기능)

2. 이벤트 스트림을 디스크에 저장함 (아파치 카프카 이전의 솔루션들과의 차이점. Write to Disk)

3. 이벤트 스트림을 분석 및 처리함

 

사용 사례

이벤트가 사용되는 모든 곳에서 사용

- 앱에서 발생하는 로그 수집

- Messaging System

- IOT 디바이스로부터 데이터 수집

- DB 동기화 (MSA 기반의 분리된 DB간 동기화)

- 실시간 ETL(데이터를 extract -> transfrom -> load)

- spark, flink, storm, hadoop 과 같은 빅데이터 기술들과 함께 사용

---

Topic, Partition, Segment

Topic: 카프카 내에서 전송되는 메시지가 저장되는 논리적인 장소. 파일이나 디렉터리처럼 눈에 보이지 않음

Producer: 메시지를 만들어서 topic으로 보내줌

Consumer: topic의 메시지를 가져와서 비즈니스에 활용

Consumer group: topic의 메시지를 사용하기 위해 협력하는 consumer들의 집합. 하나의 consumer는 하나의 consumer group에 포함되며, consumer group 내의 consumer들은 협력하여 topic의 메시지를 분산 병렬 처리함

 

producer와 consumer의 분리(decoupling)

- producer와 consumer는 서로 알지 못하며, producer와 consumer는 각각 고유의 속도로 commit log에 write 및 read를 수행함

- 다른 consumer group에 속한 consumer들은 서로 관련이 없으며, commit log 에 있는 event(message)를 동시에 다른 위치에서 read할 수 있음

 

commit log

- 추가만 가능하고 변경 불가능한 데이터 스트럭처. 데이터(event)는 항상 로그 끝에 추가되고 변경되지 않음.

- 0번에 event가 write -> 1번에 event가 write ... 

- 카프카도 이와 같은 commit log 개념 기반으로 만들어짐

- Producer가 데이터를 추가(write)하는 가장 마지막 로그를 LOG-END-OFFSET

- Consumer가 데이터를 읽(read)고 마킹(commit)하는 offset을 CURRENT-OFFSET

- LOG-END-OFFSET과 CURRENT-OFFSET 의 위치의 차이를 consumer lag이라고 부름

 

Partition: Commit Log. Topic을 구성하는 컴포넌트. 하나의 Topic은 하나의 Partition 이상으로 구성을 하도록 만들어져 있음. 다수의 Partition은 사용하는 것은 병렬처리를 하기 위함. 

Segment: Partition을 구성하고 있는 실제 물리 파일. Segment file이 지정된 크기보다 크거나 지정된 기간보다 오래되면 새 파일이 열리고 메시지는 새 파일에 추가됨

 

- Topic은 생성시 Partition개수를 지정하게 되어있음

- 각 Partition은 broker들에 분산되며 Segmenet file들로 구성됨 (예를들어 1번 Topic의 partition A는 1번 broker에, 1번 Topic의 partition B는 2번 broker에 있을 수 있음)

- 카프카 클러스터는 여러개의 broker들로 구성되어있음

- Partition당 오직 하나의 Segment만 활성화되어 있음. 데이터가 계속 쓰여지고 있는 중이기 때문. 1번 segment가 꽉차면 2번 segment에 이어서 쓰고.. 이런식임

---

Broker, Zookeeper

Kafka Broker

- Topic과 Partition을 유지 및 관리하는 소프트웨어

- Kafka Server, Bootstrap Server라고 부르기도 함

- 각각의 Broker들은 ID로 식별됨

- Topic의 일부 Partition들을 포함 (Topic에 N개의 Partition이 있을 때 여러 broker에 분산되어서 관리되기 때문. 장애 발생에 대비하기 위한 것 중 하나)

- 여러 개의 Broker들을 Kafka Cluster라고 함

- Client(Consumer, Producer 등)는 특정 Broker에 연결하면 전체 클러스터에 연결됨. 연결된 특정 Broker가 전체 Broker들의 리스트를 전달하고 Client는 접속해야할 Topic이 어디 있는지 알게돼서 필요한 Broker쪽으로 접속하게 됨. (장애를 대비해 각각의 Broker는 모든 Broker, Topic, Partition에 대해 알고 있음)

- 최대 3대 이상의 Broker들로 하나의 Cluster 구성해야 함. 4대 이상을 권장함

- Topic 생성시 Kafka가 자동으로 Topic을 구성하는 전체 Partition들을 모든 Broker에게 할당해주고 분배해 줌

 

Zookeeper

- Broker들의 목록/설정을 관리하고 변경사항(Topic 생성 제거 등)에 대해 Broker들에게 알려주는 역할을 하는 소프트웨어

- Zookeeper는 홀수의 서버로 작동하게 설계되어 있음(최소 3, 권장 5) -> Quorum 쿼럼 알고리즘 기반으로 의결을 하는게 필요한 최소한도의 인원수를 충족시키기 위함. 앙상블이 3대 있을 때 예상치 못한 장애가 발생해 1개의 서버에서 장애가 발생하더라도 2대, 즉 과반수 이상이 존재하므로 정상 동작 (쿼럼은 2)

- N개의 Zookeeper 서버: Zookeeper Ensemble 앙상블

- 대용량 분산 시스템을 위한 네이밍 레지스트리를 제공하는 소프트웨어 -> 멀티 Kafka Broker들 간의 정보 공유, 동기화를 수행하기 위해 Tree 형태의 데이터 저장소로 구성되어 있음

- Zookeeper 없애는 버전 출시됨

참고) 왜 Zookeeper를 없앴을까?
- 3.3.1 버전부터
- 카프카 외부에서 메타데이터 관리하다보니 데이터 중복, 브로커의 메타 데이터와 주키퍼의 메타데이터의 불일치, 시스템 복잡성 증가, 서버 추가 필요 등의 문제 -> 카프카 외부에서 메타데이터를 관리하는게 문제가 아닐까? -> KRaft 모드(이벤트 기반의 Quorum 쿼럼 컨트롤러 서비스를 사용)

KRaft 모드
- 메타데이터와 configuration을 @metadata라는 이름의 토픽으로 저장함
- 몇개의 선택된 서버가 controller가 되고 이것들로 내부의 쿼럼이 구성되어짐. 쿼럼이 @metadata를 관리함
- controller는 active 혹은 standby mode 둘중 하나가 되고, 만약 현재 active인 controller 서버가 탈락하면 standby mode인 서버가 active 된다고 함

https://hoing.io/archives/4029
https://medium.com/free-code-camp/in-search-of-an-understandable-consensus-algorithm-a-summary-4bc294c97e0d

---

Producer

Producer: 메시지를 생성해서 kafka의 topic으로 메시지를 보내는 애플리케이션

 

Producer가 보내는 (Message == Record == Event == Data)의 구성

Headers    -> Metadata

Key, Value -> 실제 보내고자 하는 데이터, 다양한 형태가 가능(JSON, String 등)

 

Kafka는 producer가 생성한 record를 byte array 로 저장(Serializers)

이를 원본 데이터 형태로 바꿔 Consumer에서 활용(Deserializers)

 

코드 상에서는 데이터, serializer만 세팅해서 넘기면 그 외의 과정은 kafka에서 처리(byte로 변환, 성공 및 재시도 처리 등)

Producer에서 생성된 record가 kafka에서 처리되는 과정

 

Partitioner: 메시지를 Topic의 어떤 Partition으로 보낼지 결정. 성능, 작동 방식이 다양함

---

Consumer

Consumer: Topic의 메시지를 가져와서 소비하는 애플리케이션. 각각 고유의 속도로 Commit log로부터 순서대로 read(poll)를 수행

 

Consumer Offset

- Consumer Group이 읽은 위치를 표시

- __consumer_offsets 라는 Internal Topic에서 Consumer Offset을 저장하려 관리

Consumer가 데이터를 읽어와서 offset을 저장하는 과정

Consumer Group

- 동일한 group.id로 구성된 모든 Consumer들은 하나의 Consumer group 을 형성

- partition은 항상 consumer group내의 하나의 consumer에 의해서만 사용됨

- Consumer는 주어진 topic에서 0개 이상의 많은 partition을 사용할 수 있음

- Consumer Group의 Consumer들은 작업량을 어느 정도 균등하게 분할함

- 동일한 Topic에서 consume하는 여러 Consumer Group이 있을 수 있음

 

Message Ordering(순서)

- Partition이 2개 이상인 경우 모든 메시지에 대한 전체 순서 보장 불가능 (하나의 Consumer가 여러개의 partition읽기 때문)

- Topic 1개에 Partition 1개만 만든다면 Consumer가 하나의 Partition만 읽기 때문에 순서보장이 됨 -> 하지만 처리량이 떨어짐 -> 순서보장을 해야하는 경우가 많을까? 고민 필요

- 동일한 Key를 가진 메시지는 동일한 Partition에 전달되어(Partitioner) Key 레벨의 순서 보장은 가능함 (partition에서 hash 알고리즘으로 Partition에 분배하기 때문)

- 운영중에 Partition 개수를 변경하면 -> 순서 보장 불가

key레벨의 순서보장

 

Consumer Failure

- Consumer를 rebalancing해서 fail이 난 consumer가 가져가던 partition을 다른 consumer가 가져가게됨

---

Replication

특정 Broker에서 장애가 발생한다면?

장애가 난 Broker에 있는 Partition들을 다른 Broker에서 생성하면 해결될까?

그렇다면 기존 메시지와 Offset 정보들은 버릴 것인가? -> 데이터 꼬임 -> 이를 해결하기 위한 기술이 Replication(복제)

 

Replication

- Partition을 복제하여 다른 Broker상에서 복제물(Replicas)을 만들어서 장애를 미리 대비함

- 원본은 Leader Partition, 복제본을 Follower Partition이라고 함

- Replication Factor = Leader, Follower 다 포함해서 N개 (Topic이나 Broker쪽에 세팅할 수 있음)

- Producer, Consumer는 Leader만 read, write 함

- Follower는 Leader의 Commit log에서 데이터를 가져오기 요청(Fetch Request)으로 복제

- 장애가 발생하면? -> Kafka Cluster가 새로운 Leader를 선출 -> Producer, Consumer는 바뀐 Leader에 read, write

- 하나의 Broker에만 Partition의 Leader들이 몰려있다면? -> 하나의 Broker에 Producer, Consumer가 몰림 -> 이를 방지하기 위한 옵션 -> auto.leader.rebalance.enable, leader.imbalance.check.interval.seconds, leader.imbalnce.per.broker.percentage

- Rack Awareness: Broker를 저장할 때 Leader, Follower 간에 최대한 Rack별로 분산하여 Rack 장애를 대비하는 옵션

 

---

In-Sync Replicas(ISR)

In-Sync Replicas

- 정말 잘 복제해가고 있는지를 판단하는 지표

- Leader가 장애가 났을 때 어떤 걸 Leader로 선출해야할지 고를때 사용됨 -> ISR이면 다음 Leader 후보가 됨

- High Water Mark(가장 최근에 Committed 메시지의 Offset 추적) 라고 하는 지점까지 동일하게 복제해간 Replicas(Leader, Follwer 모두)의 목록

- replica.lag.max.messages: 해당 옵션이 4일 때 아래의 Broker103에서 Leader와 4이상으로 차이나므로 ISR이 아니고 OSR이라고 부르게 됨 -> 메시지 유입량이 갑자기 늘어나는 경우 지연으로 판단될 수 있음 -> replica.lag.time.max.ms로 판단해야 함

- replica.lag.time.max.ms: Follwer가 Leader로 fetch 요청을 보내는 Interval을 체크. 해당 옵션안에 fetch 요청이 온다면 잘 따라잡고 있구나-라고 판단

- ISR은 해당 Leader가 떠있는 Broker가 관리함

ISR 개념

 

Controller

- Kafka Cluster 내의 Broker 중 하나가 Controller가 됨

- Zookeeper를 통해 Broker Liveness를 모니터링. Leader와 Replica 정보를 Cluster 내의 다른 Broker들에게 전달(캐싱해서 속도 높이기 위해)

- Leader 장애시 Leader Election(선출)을 수행함

- Controller가 장애시 다른 Active Broker들 중에서 재선출

 

Consumer 관련 Position들

- Last Commited Offset(Current Offset): Consumer가 최종 Commit한 Offset

- Current Position: Consumer가 읽어간 위치(처리 중, Commit 전)

- High Water Mark (Committed): ISR(Leader-Follower)간에 복제된 Offset

- Log End Offset: Producer가 메시지를 보내서 저장된, 로그의 맨 끝 Offset

- Consumer Lag: Log End Offset과 Last Commited Offset의 차이

 

Committed의 의미

- ISR 목록의 모든 Replicas가 메시지를 성공적으로 가져오면 Committed

- Consumer는 Committed 메시지만 읽을 수 있음

- Leader는 메시지를 Commit 할 시기를 결정

- Committed 메시지는 모든 Follwer에서 동일한 Offset을 갖도록 보장

- 어떤 Replica가 Leader인지에 관계없이 모든 Consumer는 해당 Offset에서 같은 데이터를 볼 수 있음

 

Leader Epoch

- 새 Leader가 선출된 시점을 Offset으로 표시

- Leader-epoch-checkpoint 파일에 체크포인트를 기록

---

Reference

출처: 패스트캠퍼스 Kafka 완전 정복: 클러스터 구축부터 MSA 환경 활용까지