Kafka 란?

Apache Kafka는 실시간으로 기록 스트림을 게시, 구독, 저장 및 처리할 수 있는 분산형 데이터 스트리밍 플랫폼.

빅데이터의 파이프라인 으로서 사용할 수 있는 데이터 스트리밍 플랫폼이다.

• 높은 처리량. 많은 양의 데이터를 묶음 단위로 처리하는 배치 처리가 가능. 또한 파티션 단위를 통해 동일 목적의 데이터를 여러 파티션에 분배하고 데이터를 병렬 처리할 수 있다.

• 확정성. 카프카 클러스터의 브로커 개수를 자연스럽게 스케일 아웃할 수 있다. 다운타임 없이 확장 가능 ( 데이터의 양을 예측하기 어려울 때 사용 하기 좋다 )

• 영속성. 데이터를 메모리에 저장하지 않고 파일 시스템에 저장한다. 페이지 캐시 메모리 영역을 사용한다! 한 번 읽은 파일 내용을 OS가 사용하는 메모리에 저장한다. 따라서 파일시스템을 사용하더라도 처리량이 높다.

• 고가용성. 3개 이상의 서버들로 운영되는 카프카 클러스터 3개 이상의 브로커를 사용한다.

카프카는 어떻게 파일로 데이터를 관리하면서도 높은 처리량을 보여줄 수 있을까?

카프카를 구성하는 요소들

주키퍼 (zookeeper)

Kafka의 3.0.0 버전이 나오면서 주키퍼가 없어도 kafka 클러스터를 구성할 수 있지만 아직은 불완전하기 때문에 주키퍼를 이용해서 관리해줘야한다. 카프카 클러스터에 있는 다수의 브로커에 대한 코디네이션을 이 주키퍼가 해주게 된다.

브로커 ( Broker )

실질적으로 데이터를 저장하고 있는 서버가 브로커이다. Producer, Consumer는 이 브로커를 통해 데이터를 저장 및 로드하게 되는 것.

Kafka 클러스터에서는 3대이상의 브로커를 기본으로 사용하여 구성되게 된다.

브로커의 역할

1. 컨트롤러

클러스터의 다수 브로커중 한 대가 컨트롤러의 역할을 한다. 컨트롤러는 다른 브로커들의 상태를 체크하고 브로커가 클러스터에서 빠지는 경우 해당 브로커에 존재하는 리더 파티션을 재분배한다. 컨트롤러의 역할을 하는 브로커에 장애가 생기면 다른 브로커가 컨트롤러 역할을 한다.

2. 데이터 삭제 & 압축

컨슈머가 데이터를 가져가더라도 토픽의 데이터는 삭제되지 않는다. 컨슈머나 프로듀서가 데이터 삭제를 요청할 수도 없다. 오직 브로커만이 데이터를 삭제할 수 있다. 데이터 삭제는 파일 단위로 이루어지는데 이 단위를 '로그 세그먼트' 라고 부른다. 이 세그먼트에는 다수의 데이터가 들어 있기 때문에 일반적인 데이터베이스처럼 선별이 불가능하다. cleanup.policy 정책에 따라서 삭제, 압축등을 할 수 있다.

• delete: Active 로그 세그먼트를 제외한 대상이되는 세그먼트를 모두 삭제하게 된다.
• compact: 토픽 압축 정책은 일반적으로 생각하는 zip 과 같은 압축과는 다른 개념이다. 메시지 키 별로 해당 메시지 키의 레코드 중 오래된 데이터를 삭제하는 정책! 일부 레코드만 제거가 가능. 액티브 세그먼트는 제외된다.

압축정책을 사용하게 될 때 테일영역과 헤드영역에 대해서 알아둬야 압축을 어떤 방식으로 진행할지 설정할 수 있다.

• 테일 영역: 압축 정책에 의해 압축이 완료된 레코드를, 클린로그라고도 부른다. 중복 메시지가 없다.
• 헤드 영역: 압축 정책이 되기 전 레코드들, 더티로그 라고도 부른다. 중복된 메시지 키가 있다.

cleanup.policy에 대한 configuration 설정방법 예시

• cleanup.policy=delete | compact 로 설정

• retention.ms(minutes, hours): 세그먼트를 보유할 최대 기간. 기본 값은 7일

• retention.bytes: 파티션당 로그 적재 바이트 값. 기본 값은 -1 ( 지정하지 않음)

• log.retention.check.interval.ms: 세그먼트가 삭제 영역에 들어왔는지 확인하는 간격. 기본 값은 5분

• log.min.cleanable.dirty.ratio # 액티브 세그먼트를 제외하고 남아있는 세그먼트에 남아 있는 테일 영역의 레코드 개수와 헤드영역의 레코드 개수의 비율을 뜻한다! trade off 가 있다. 자주 하면 브로커에 부담 크게 하면 효율적.

3. 컨슈머 오프셋 저장

__consumer_offsets라는 토픽에 커밋한 오프셋을 저장한다. 여기에 저장된 오프셋을 토대로 컨슈머 그룹은 다음 레코드를 가져가서 처리한다.

4. 그룹 코디네이터

컨슈머 그룹의 상태를 체크하고 파티션을 컨슈머와 매칭되도록 분배하는 역할을 한다. 리밸런스 작업!

5. 데이터 저장

카프카를 실행할 때 config/server.properties의 log.dir 옵션에 정의한 디렉토리에 데이터를 저장. 토픽의 이름과 파티션 번호의 조합으로 하위 디렉토리를 생성하여 데이터를 저장한다.

log.dirs=/Users/me/Documents/test/kafka_2.12-2.5.0/data # 와 같이 정의했다면

# 아래와 같이 토픽과 파티션 조합으로 형성된다.
/Users/me/Documents/test/kafka_2.12-2.5.0/data/topic_name-0
/Users/me/Documents/test/kafka_2.12-2.5.0/data/topic_name-1
.
.
.
/Users/me/Documents/test/kafka_2.12-2.5.0/data/topic_name-5

log 에는 메시지와 메타데이터를 저장한다. index는 메시지의 오프셋을 인덱싱한 정보를 담은 파일이다. timeindex 파일에는 메시지(레코드)에 포함된 timestamp값을 기준으로 인덱싱한 정보가 담겨 있다.

# 해당 폴더들에는 다음과 같은 파일들이 존재
00000000000000000000.index 00000000000000000000.log
00000000000000000000.timeindex leader-epoch-checkpoint

브로커에 데이터가 저장될 때는 하나의 로그 세그먼트 파일에 모두 담기는게 아닌 byte 기준 혹은 시간 기준으로 파일을 나누어서 저장하게 된다.

00000000000000000100.log
00000000000000000200.log

해당 주기 혹은 기준이되는 byte크기는 configuration을 통해 수정이 가능하다.

log.segment.bytes: 바이트 단위의 최대 세그먼트 크기 지정. 기본 값은 1GB

log.roll.ms(hours): 세그먼트가 신규 생성된 이후 다음 파일로 넘어가는 시간 주기. 기본 값은 7일

가장 마지막 세그먼트 파일을 액티브 세그먼트라고 부른다. 액티브 세그먼트는 브로커의 삭제 대상에서 포함되지 않는다. 액티브 세그먼트가 아닌 세그먼트는 retention 옵션에 따라 삭제 대상으로 지정된다.

6. 데이터 복제

카프카를 장애 허용 시스템으로 동작하도록 하는 원동력. 복제의 이유는 브로커 중 일부에 장애가 발생하더라도 데이터를 유실하지 않고 안전하게 사용하도록 하기 위함.

복제된 파티션은 리더(leader)와 팔로워(follower)로 구성된다. 프로듀서 또는 컨슈머와 직접 통신하는 파티션을 리더, 나머지를 팔로워라고 한다. 팔로워 파티션은 리더 파티션의 오프셋을 확인하여 현재 자신이 가지고 있는 오프셋과 차이가 나는 경우 리더 파티션으로 부터 데이터를 가져와 자신의 파티션에 저장한다.

디스크용량을 많이 차지하는 단점이 있지만 안정성이 중요하기 때문에 감수해야함...! 최소 2개 이상의 복제개수를 해야함.

리더 파티션에서 장애가 발생하면 팔로워 파티션이 리더로 승급되게 된다.

ISR(In-Sync-Replicas)

리더 파티션, 팔로워 파티션의 오프셋이 일치하여 싱크가 맞음을 뜻한다. 복제가 완료되지 않은 상태에서 리더 파티션이 장애가 발생하게 되면 팔로워가 승급했을 때 데이터 유실이 일어날 수 있다.

-unclean.leader.election.enable=true : 유실을 감수함.

-unclean.leader.election.enable=false : 유실을 감수 x

토픽 단위로 설정이 가능하다!

---

토픽과 파티션

토픽은 카프카에서 데이터를 구분하기 위해 사용하는 단위이다.토픽은 1개 이상의 파티션을 소유하고 있다. 파티션에는 프로듀서가 보낸 데이터들이 들어가 저장되는데 이 데이터를 레코드라고 부른다. 파티션은 자료구조의 큐와 비슷한 구조이다.

파티션이 5개인 토픽을 생성했을 경우 0번 브로커부터 round-robin 방식으로 리더 파티션들이 생성된다.
이를 통해서 하나의 브로커에 통신이 집중되는 상황을 막음! ( Client와의 통신은 리더 파티션만 하기 때문이다.)

참고로 파티션은 한 번 형성되게 되면 줄이는 것이 불가능하다 파티션이 사라짐에 따른 데이터들의 재분배가 일어나야하는데 그로인해 브로커에 상당한 부하가 걸리기 때문에 막아 놓은듯. 따라서 잘 생각해서 확장해야한다.

레코드

타임스탬프, 헤더, 메시지 키, 메시지 값, 오프셋 으로 구성되어 있다. 한 번 적재된 레코드는 수정 불가하다.

타임스탬프

스트림 프로세싱에서 활용!
프로듀서에서 따로 설정하지 않으면 ProducerRecord 생성 시간 이 들어간다. 또는 브로커 적재 시간 (LogAppendTime)으로 설정할 수도 있다. 토핑단위로 설정이 가능하다.

오프셋

프로듀서가 생성한 레코드에는 없다. 브로커에 적재될 때 형성이됨.

브로커에 적재될 때 오프셋이 지정됨 0부터시작해서 1씩증가 컨슈머는 오프셋을 기반으로 처리가 완료된 데이터와 앞으로 처리해야할 데이터를 구분 파티션별로 고유한 오프셋이 있다.

헤더

key/value 데이터를 추가할 수 있다.

레코드의 스키마 버전이나 포맷과 같이 데이터 프로세싱에 참고할 정보들을 담는다.

메시지 키

메시지 값의 분류를 위해 사용됨. 이를 파티셔닝이라고 한다.프로듀서가 브로커에 메시지를 보낼 때 파티셔너가 어느 파티션에 들어갈지 처리해준다. 필수는 아니고 지정하지 않으면 null로 처리 null인 레코드는 특정 토픽의 파티셔닝에 라운드 로빈 방식으로 전달된다. null이 아닌 메시지 키는 해쉬값에 의해서 특정 파티션에 매핑되어 전달된다.

키에 따른 해쉬값으로 파티션이 정해지기 때문에 이에 따른 순서보장이 가능하게 된다. ( 다른 파티션에 들어가 처리될 일이 없기 떄문 )

메시지 값

실질적으로 처리할 데이터가 담기는 공간 ( csv, json, object ) 다양한 형태로 지정가능 직렬화/역직렬화 클래스를 만들 수도 있음 컨슈머에서는 역직렬화 포맷을 알고 있어야함!!! ( 보통 string으로 직렬화 역직렬화 공간 낭비가 있음 )

Reference

도서 : 아파치 카프카 애플리케이션 개발을 위한 「실전 가이드」