☕ 개요
이번 글에서는 Kafka에 대한 개념적인 내용들을 좀 더 자세히 알아보도록 하겠다.
이 글은 Dev원영님의 강의를 듣고 정리한 내용이다. 🙏
🦴 카프카의 뼈대: 브로커, 클러스터, 주키퍼
주키퍼 (Zoo keeper)
- 주키퍼는 분산 코디네이션 서비스이고, 카프카 클러스터를 실행하기 위해서 사용되는 애플리케이션이다.
- 카프카 3.x 버전 이상부터는 주키퍼가 없어도 카프카 클러스터를 실행할 수 있다. 하지만, 아직은 주키퍼를 100% 대체하지는 못하고 있어서, 주키퍼가 여전히 많이 사용되고 있다.
브로커와 클러스터
- 브로커는 카프카 클라이언트와 데이터를 주고 받는 주체이다.
- 하나의 서버에는 하나의 브로커 프로세스가 실행된다.
- 일반적으로 데이터를 안전하게 스트리밍하기 위해서 3개정도의 브로커를 묶어서 하나의 클러스터로 운영한다. (하나의 클러스터에 속할 수 있는 브로커의 갯수는 제한은 없다.)
🧐 브로커의 역할
1) 컨트롤러
- 하나의 클러스터의 여러 브로커 중에서 한 대가 "컨트롤러"역할을 수행한다
- 컨트롤러는 다른 브로커들의 상태를 체크하고, 장애가 발생한 브로커를 클러스터에서 제외하고, 리더(Leader) 파티션을 재분배하는 역할을 한다.
2) 데이터 삭제
- 카프카는 컨슈머가 데이터를 가져가도, 데이터가 삭제되지 않는다. 오직 브로커만이 데이터를 삭제할 수 있다.
- 데이터 삭제는 파일 단위로 이루어진다. 이 "단위"를 "로그 세그먼트"라고 부른다. 로그 세그먼트는 일반적인 데이터베이스처럼 특정 데이터를 선별해서 삭제할 수는 없다.
3) 컨슈머 오프셋 저장
- 컨슈머가 특정 파티션으로부터 데이터를 가져가면, 이 파티션의 몇번째 레코드까지 가져갔는지 확인하기 위해서 "오프셋"을 "커밋"한다.
- 커밋된 오프셋은
__consumer__offsets토픽에 저장된다.
4) 그룹 코니테이터
- "코디네이터"는 컨슈머 그룹의 상태를 체크하고, 파티션을 컨슈머와 매칭되도록 분배한다.
- 컨슈머가 컨슈머 그룹에서 빠지면, 매칭되지 않은 파티션을 정상 동작하는 컨슈머로 할당하여 끊임없이 데이터가 처리되도록 돕는다. 이러한 과정을 "리밸런스"라고 부른다.
5) 데이터 저장 1 : 기본
config/server.properties의log.dir옵션에 정의된 디렉토리에 데이터를 저장한다.- 토픽이름과 파티션 번호의 조합으로 하위 디렉토리를 생성하여 데이터를 저장한다.
xxx.log: 메세지와 메타데이터가 저장되는 파일.xxx.index: 메세지의 오프셋을 인덱싱한 정보가 저장된다.xxx.timeindex: 메세지에 포함된timestamp값을 기준으로 인덱싱항 정보가 저장된다.
6) 데이터 저장 2 : 세그먼트
- 데이터는 시간과 데이터의 크기에 따라서 파일로 구분하여 저장한다.
log.segment.bytes: 하나의.log파일에 저장할 최대 세그먼트 크기를 지정할 수 있는 옵션이다. (Default: 1GB)log.roll.ms: 데이터를 새로운.log파일에 저장하는 시간 주기 (Default: 7일)- 가장 마지막에 생성된 세그먼트를 "Active 세그먼트"라고 부른다. 현재 데이터가 계속 쓰여지고 있는 파일인 것이다.
🔥 데이터 삭제 주기
데이터의 삭제는 크게 2가지 옵션으로 구분이 된다.
1) cleanup.policy=delete
정해진 기준을 넘어가는 경우 파일을 삭제하는 옵션이다. 여기서 정해진 기준은 아래의 옵션들로 결정이 된다.
retention.ms: 세그먼트를 보유할 최대 기간 (Default 7일)retention.bytes: 파티션당 적재할 로그의 용량 {Default -1 (지정하지 않음을 뜻 함)}log.retention.check.interval.ms: 세그먼트가 삭제의 대상이 되는지 확인하는 주기. (Default: 5분)
2) cleanup.policy=compact
정해진 기준을 넘어가는 경우 파일을 압축하는 옵션이다. 여기서 압축한다는 것은 동일한 메세지 Key를 갖고 있는 레코드 중에서 가장 최신 레코드들만 남기고 나머지 레코드들은 삭제하는 것을 의미한다. (단, Active 세그먼트는 삭제 대상에서 제외된다.)
- 압축 정책에 의해서 압축된 레코드들을 "Tail 영역" 또는 "Clean 로그"라고 부른다.
- 아직 압축되지 않은 레코드들은 "Head 영역" 또는 "Dirty 로그"라고 부른다.
min.cleanable.dirty.ratio: 클린 레코드와 더티 레코드의 상대적인 비율(클린/더티)을 뜻하고, 이 옵션보다 비율이 커지면, 레코드 압축이 수행된다.
이 값을 너무 작게 설정하면, 압축 작업이 너무 빈번하게 일어나서 문제이고, 반대로 너무 크게 설정해도, 압축이 수행되기 전까지 부담해야하는 데이터 용량이 너무 커질 수 있다. 즉, 적절하게 설정해야 한다. 🤓
👯 데이터 복제
클러스터로 묶인 브로커 중 일부에서 장애가 발생하더라도, 데이터를 유실하지 않고 안전하게 데이터 스트리밍을 하기 위한 장치이다.
- 데이터 복제는 "파티션" 단위로 이루어진다.
- 토픽을 생성할 때, 파티션의 복제 개수(Replication factor)도 같이 설정된다. 따로 설정하지 않으면, 브로커에 설정된 옵션을 따라간다.
- Replication factor의 최대값은 "브로커의 개수"와 같다.
리더 파티션
- 프로듀서, 컨슈머와 직접 통신하는 파티션이다.
팔로워 파티션
- 리더 파티션의 Offset을 확인하여, 자신의 Offset과 차이가 나는 경우 리더 파티션으로 부터 데이터를 가져온다. (Replication)
- 리더 파티션이 속해있는 브로커에서 장애가 발생하면, 팔로워 파티션 중 하나가 리더 파티션이 된다. 이러한 과정을 통해서 데이터 유실을 방지할 수 있다.
- 토픽별로 복제 개수를 다르게 설정할 수 있다. 데이터 유실이 치명적인 토픽이라면, 복제 개수를 3이상으로 설정하는 것이 바람직하다.
ISR(In-Sync Replica)와 장애 대처
ISR이란, 리더 파티션과 팔로워 파티션이 모두 싱크되어 있는 것을 뜻한다. 여기서 "싱크"되어 있다는 것은 리더 파티션과 팔로워 파티션의 Offset이 동일하다는 것을 의미한다.
반대로 ISR이 되지 않았다면, 데이터의 복제가 일어나지 않았거나 진행중인 상황인 것이다. 이러한 상황에서 리더 파티션이 속한 브로커에서 장애가 발생한다면, 팔로워 파티션에 복제되지 않은 데이터를 어떻게 처리할 것인지 결정해야한다.
unclean.leader.election.enable=true: 복제되지 않은 데이터들의 유실을 감수하고, 팔로워 파티션 중 하나를 리더로 승격시킨다.unclean.leader.election.enable=false: 장애가 발생한 리더 브로커가 복구될 때까지 대기해서 데이터 유실을 방지한다. 단, 복구될 때까지 대기하는 동안에는 리더 파티션이 속한 토픽은 동작하지 않는다.
✍️ 토픽과 파티션
- 카프카 클러스터는 카프카 브로커들로 구성이 되고, 하나의 브로커 안에는 N개의 토픽들이 저장된다.
- 토픽은 데이터를 구분하기 위해서 사용하는 단위이고, 1개 이상의 파티션을 소유하게 된다.
- 각 파티션에는 프로듀서가 전송한 데이터들이 저장되고, 이를 "레코드(Record)"라고 부른다.
- 파티션은 FIFO 구조이다.
파티션 생성 규칙
파티션이 N개인 어떤 토픽을 생성했을 때, Round-robin 방식을 통해서 브로커들에 골고루 리더 파티션이 생성되게 된다.
리더 파티션이 생성되지 않은 브로커에는 팔로워 파티션이 생성된다.
파티션 쏠림 현상
카프카 내부적으로 파티션을 골고루 분배하긴 하지만, 혹시나 리더 파티션이 특정 브로커에만 너무 많이 할당이 된다면, kafka-reassign-partitions.sh를 실행해서 파티션을 재분배할 수 있다.
파티션, 컨슈머 개수와 처리량
파티션 1개는 컨슈머 1개와 매칭이 된다. 그래서 처리량을 늘리기 위해서는 컨슈머의 개수를 늘리고, 그에 대응하도록 파티션의 개수도 늘려주어야한다.
주의할 점은 파티션의 개수는 한번 늘리면 다시 줄일수는 없으므로, 신중하게 파티션의 개수를 설정해야한다.
토픽 이름 정하기
카프카에서는 토픽의 이름은 한번 지정하면, 수정할 수 없으므로 잘 지어주어야한다.
아래의 패턴들로 이름을 짓는 것이 일반적이다.
{환경}.{팀 이름}.{애플리케이션 이름}.{메세지 타입}
예시)prod.engineer-team.core-server.json{프로젝트 이름}.{서비스 이름}.{환경}.{이벤트 이름}
예시)batch-server.hello.prod.notification{환경}.{서비스 이름}.{JIRA 번호}.{메세지 타입}
예시)prod.hello.jira-1133.csv{클러스터 이름}.{환경}.{서비스 이름}.{메세지 타입}
예시)aws-kafka-1.prod.hello.json
💽 레코드
하나의 레코드는 이렇게 구성이된다.
timestamp
- 데이터 스트리밍에 활용하기 위한 "시간"을 저장한다.
- 아무 값도 지정하지 않으면, 레코드가 생성된 시간이나 브로커에 적재된 시간이 입력된다.
- 이 옵션은
message.timestamp.type으로 설정할 수 있다.
offset
- 브로커에 레코드가 적재될 때, 입력되는 값이다.
- 오프셋은 0부터 시작해서 1씩 증가하는 특징을 갖고 있다.
- 컨슈머들은 내가 어떤 오프셋까지 처리했는지를 기록한다. 즉, 이를 통해서 다음 작업을 이어서 하거나 중복 작업을 방지할 수 있다.
headers
- Key/Value 형태의 데이터를 저장할 수 있다.
- 레코드의 스키마 버전이나 포맷과 같은 메타 데이터를 담고 있다. Http 통신의 헤더와 유사한 성격이다.
key
- 메세지 분류를 위해서 사용되는 값이다.
- key값을 해싱하여, 이에 대응하는 파티션에 레코드가 할당되게 된다. 동일한 key값이라면, 무조건 동일한 파티션에 할당되게 된다.
- key가
null이라면, 파티션들에 랜덤하게 골고루 들어가게 된다.
value
- 실질적인 데이터를 의미한다.
- Float, Byte[], String, CSV, JSON등등 다양한 데이터를 입력할 수 있다.
🤓 카프카 클라이언트 (프로듀서, 컨슈머)
클라이언트가 통신하는 법
- 카프카 클라이언트는 카프카 클러스터에 "메타데이터"를 요청한다.
- 메타 데이터 안에는 "리더 파티션"의 위치가 담겨 있다.
- 카프카 클라이언트는 리더 파티션과 직접 통신하여 데이터를 송/수신한다.
프로듀서의 메타데이터 옵션
metadata.max.age.ms: 메타 데이터를 강제로 갱신하는 간격 (Default: 5분)metadata.max.idle.ms: 메타 데이터의 캐시를 유지하는 시간 (Default: 5분)
