아파치 카프카 애플래케이션 프로그래밍 with 자바

기존 1대 1 데이터 파이프라인은 장애가 전파되지만 카프카로는 끊을 수 있다.

데이터 포맷에는 제한이 거의 없다.

상용환경은 최소 3대의 브로커를 운영한다

왜 매직넘버3을 사용할까

1대 : 장애가 나면 서비스의 장애로 이어진다

2대 : 한 대가 죽을 경우 다른 한대가 리더가 되지만 복제 시간차로 인해 메시지 유실이 있을 수 있다

리플리카 2 설정

3대 : 2대가 복제가 돼야 메시지를 안정적으로 처리했다고 판단한다

2대 : 2가 설정 값이기 때문에 한 대가 죽으면 브로커는 메시지를 받지 않아서 서비스 장애

 

카프카는 데이터를  보낼 때와 소비할 때 모두 배치 처리하기에 효율적이다. 

브로커를 늘리거나 줄임으로써 효율적으로 관리할 수 있다.

카프카는 메시지를 파일단위로 관리하는데 느릴 거 같지만 운영체제 레벨에서 파일 IO를 사용하기에 빠르다.

페이지 캐시 영역을 메모리에 따로 생성하여 사용한다.

 

데이터 람다 아키텍처는 스피드, 배치, 서빙 레이어로 나뉘는데 카프카는 스피드 레이어에 위치한다.

배치와 스피드가 중복 데이터처리를 하기 때문에 비효율이 있다. -> 카파 아키텍처는 모두 스피드 레이어를 통해 간다

배치 데이터: 지난 1분간의 주문데이터, 스트림데이터: 클릭로그 등.

kafka는 jvm위에서 돌아간다. 레코드의 내용은 페이지 캐시로 시스템 메모리를 사용하고 나머지 객체들은 힙 메모리에 저장하여 사용한다. 기본 설정은 브로커 1G, 주피터 1G 메모리이다.

 

커맨드 손에 익히기

 

주키퍼 실행

bin 주키퍼스타트 -데몬 컨피그/주키퍼프로퍼티

카프카 실행

bin 카프카스타트 -데몬 컨피그 서버브로퍼티

카프카 토픽 기본 생성

bin 카프카토팍 --크리에이트 --부트스트랩서버 ~~:9092 --topic 토픽이름 부트스트랩 = 클러스터를 유지할 카프카 브로커들을 명시한다

토픽 복합 설정

bin 카프카토픽 --크리에이트 --bootstrap-server localhost:9092 --partitions 3 --relication-factor 1 --config retention.ms=1728000000 --topic hello2.kafka

리플릭카 팩터의 최대 수는 통신하는 브로커의 수
파티션의 데이터는 브로커마다 저장된다
기본 설정은 config 밑에 있는 서버프로퍼티를 따라간다. 추가 설정이 가능하다 리텐션은 데이터 보존기간.
대부분 실 운영에서의 브로커는 3대이며 relication-factor는 2~3을 유지한다.

토픽확인

bin/kafka-topic.sh --bootstrap-server localhost:9092 --list bin/kafka-topic.sh --bootstrap-server localhost:9092 --describe --topic 토픽이름

파티션 별로 리더와 팔로워가 있음

설정 변경

설정은 topic.sh와 config.sh를 통해 변경할 수 있음 파티션의 수는 topic.sh에서, 삭제, 리텐션 변경은 config.sh를 사용해야 함 두 곳에서 관리되는 이유는 원래는 topic에 있었다가 이동 중 이기 때문이다.

bin/kafka-topics.sh --bootstrap-server localhost:9092 --topic hello2.kafka --alter --partitions 4./bin/kafka-configs.sh --bootstrap-server localhost:9092 --entity-type topics --entity-name hello2.kafka --alter --add-config retention.ms=86400000

프로듀서

토픽에 넣는 데이터는 레코드라고 한다. 키 벨류로 되어있다.

순서를 보장하고 싶다면 파티션을 하나만.

컨슈머 그룹

컨슈머를 동작할 때 그룹이름을 지정하면 새로 생성된다. bin kafka-consumer-groups.sh --bootstrap-server localhost:9092 --list bin kafka-consumer-group.sh --bootstrap-sever localhost:9092 --group 그룹이름 --describe

--bootstrap-server localhost:9092 = 클러스터 정보는 계속 적어준다.

컨슈머의 렉이 증가한다? 컨슈머의 처리량이 프로듀서의 메시지 생산량보다 느리다.

verifiable로 네트워크 통신 테스트를 할 수 있다.

./bin/kafka-verifiable-producer.sh --bootstrap-server localhost:9092 --max-message 10 --topic hello2.kafka 

./bin/kafka-verifiable-consumer.sh --bootstrap-server localhost:9092 --topic hello2.kafka --group-id hello-gr oup 

 

 

카프카 클로스터로 묶인 브로커들은 안전하게 저장되고 복제된다

레코드를 파일에 저장해서 느릴 거 같지만 페이지 캐시를 사용하기에 성능이 나온다. 페이지 캐시란 OS가 파일 IO 성능향상을 목적으로 만들어 놓은 메모리. 동일한 파일에 접근하면 메모리에 있는 정보를 읽는다.

장애허용시스템.

복제는 파티션 단위이다. 복제 설정은 명시하지 않으면 기본 설정 값을 쓰며 최대 수는 브로커 수이다.

복제는 팔로워가 리더의 오프셋을 확인하고 이뤄진다.

브로커 중 한대는 컨트롤러이다. 리더 선출 등을 한다.

데이터 삭제는 브로커만 가능하고 삭제 정책 말고도 압축을 하는 정책도 있다.

브로커 중 한대는 코디네이터 역할을 하며 컨슈머 그룹을 체크하고 파티션과 컨슈머를 매칭한다. 리밸런싱 시 역할을 수행한다.

주키퍼는 카프카의 메타데이터를 저장한다.

 

카프카 클러스터로 묶인 브로커들은 동일한 경로의 주키퍼 경로를 설정해야 한다.

성능을 올리려면 파티션과 컨슈머를 늘리자.

작명을 신중하게.

저장 단위를 레코드라고 하는데 레코드는 키, 값, 오프셋, 타임스탬프,  헤더를 가지고 있다.

키는 순서대로 처리 OR 메시지의 값의 종류를 나타내기 위해 사용된다. 키 값을 해시로 만들어 동일한 파티션에 적재할 수 있는데 리밸런싱이 이뤄진다면 이전과 동일한 파티션을 보장하지 않는다.

프로듀서가 메시지를 직렬화하여 브로커에 전달한다. 컨슈머는 동일한 직렬화전략을 사용해야 한다.

오프셋은 브로커가 last + 1 형태로 사용한다.

브로커에 송수신 명령을 내리는 카프카 클라이언트(프로듀서, 컨슈머, 어드민 클라이언트)

 

프로듀서는 브로커로 데이터를 보내기 전에 내부적으로 파티셔너->배치생성 단계를 거친다.

파티셔너의 역할은 토픽의 어느 파티션으로 보낼지 정하는 것이다.

디폴트인 유니폼스티키파티셔너는 배치를 기다렸다가 모두 동일한 파티션으로 보내기에 성능이 좋다.

배치 처리를 위한 데이터는 어디에 쌓이는가? 어큐물레이터에 버퍼로 쌓아놓는다.

샌더 스레드가 버퍼에 쌓인 것을 브로커에 전송한다.

ack 설정은 프로로듀서가 브로커에게 정상적으로 메시지를 받았는지 알 수 있는 설정이다.

컨슈머 운영 방법1. 하나이상의 컨슈머로 이루어진 컨슈머 그룹 운영, 2. 특정 파티션만 구독하는 컨슈머 운영

 

1개의 파티션은 최대 1개의 컨슈머에 할당 가능하다.

1개의 컨슈머는 여러개의 파티션을 할당받을 수 있다.

-> 컨슈머의 수는 파티션 수 보다 작거나 같아야 리소스를 아낄 수 있다.

컨슈머 그룹 운영ex. cup나 메모리 사용량을 하둡이나 일라스틱 서치에 저장한다고 할 때 동기 적으로 하면 외부 장애 시 적재가 불가능 하니 중간에 카프카를 두고 각각 그룹으로 운영한다면 장애가 나도 나중에 그룹단위로 각자 읽으면 된다.

 

브로커 중 한대가 그룹 코디네이터를 하는데 이는 리밸런싱을 일으키는 존재이다. 

컨슈머는 정해진 시간마다 하트비트를 보내야 하며 설정 시간이 지나도 보내지 않는다면 코디네이터는 해당 컨슈머를 제외시키고 파티션을 컨슈머에게 재할당하는 리밸런싱을 한다.

리밸런싱으로 인해 메시지 중복, 유실이 발생할 수 있으니 대응코드를 작성해 두자.

 

오토커밋(기본 5초)을 통해  poll이후에 어디까지 읽었는지 브로커의 기본 토픽에 남길 수 있는데 5초 이전에 리밸런싱과 컨슈머 강제 종료가 이뤄진다면 메시지 유실 혹은 중복이 이뤄질 수 있다.

중복이나 유실이 있어서는 안 된다면 오토커밋은 사용하지 말자.

직접 커밋은 안정성이 높다/

동기 : 성능은 떨어지지만 직접 커밋을 확인하기 때문에 안정성이 높다

비동기 : 비동기 커밋 중 오류가 발생했다면 메시지가 중복 처리될 수 있다.->순서보장 X

 

데이터의 중복과 유실이 이뤄지는 케이스는 무엇일까?

중복 :  컨슈머가 커밋을 정상동작했는지 확인하지 않았을 때 만약 커밋이 비정상적이라면 나중에 다시 데이터를 poll 할 것이다, poll이후 커밋을 보내기 전에 리밸런싱 이뤄져 파티션을 알 수 없을 때, 오토커밋을 사용하지 않고 비동기 커밋을 할 때 실패할 경우(근데 응답값을 보고 다시 롤백하면 되긴 하지만 비동기 처리의 이유인 성능을 떨어뜨린다. 안정성을 원하면 동기로 하던가),

프로듀서의 경우에서는 ack=0이 아닐 때 발생할 수 있다. 이벤틀르 발행했지만 네트워크 문제로 ack전달이 늦을 때 프로듀서는 설정 시간을 초과하면 실패로 간주하고 재발행을 하는데 이때 중복이 발생할 수 있음. 

 

(거의 커밋을 보내기 전에 트랜잭션으로 묶을 수 없는 저장소에 데이터를 저장완료했을 경우,)

유실 (거의 프로듀서 장애):  커밋을 보내고 컨슈머가 데이터 처리에 실패했을 때, 팔로워 파티션이 리더 파티션을 복제하지 못했는데 프로듀서가 ack=1 설정으로 인해 완료처리되었을 때 

(거의 프로듀서가 메시지를 보냈다고 판단했지만 브로커가 데이터를 저장에 실패했을 때

컨슈머가 완료했다고 브로커에 커밋을 했지만 데이터 저장에 실패했을 때)

 

 

자동 커밋은 poll 기본설정 5 초이후에 발생하기에 리밸런싱 대응 코드가 있다면 중복을 회피할 수 있다.

훅으로 리밸런싱 직전에 마지막 커밋을 직접 남기면 된다.

 

카프카 스트림즈는 토픽에 적재된 데이터를 상태기반 또는 비상태기반으로 실시간 변환하여 다른 토픽에 적재하는 라이브러리이다.

메시지 키를 사용하고 순서보장을 원한다면 파티션의 변화가 없도록 운영해야 한다.

파티션은 파일 시스템을 사용하기 때문에 너무 많이 늘리면 OS는 프로세스마다 파일 시스템 제한을 두고 있다.

세그먼트는 토픽의 데이터를 저장하는 명시적인 시스템 단위.

 

In-sync-replicas는 리더파티션과 팔로워파티션이 모두 싱크 된 상태를 의미

팔로워가 리더를 복제하는데 시간이 걸린다. 리더는 파티션이 잘 복제했는지 체크하는데 복제 못하고 있으면 ISR 그룹에서 제외한다.

ISR그룹에 속하지 않아도 설정으로 리더로 선출될 수 있다. 리더가 다시 살아나기를 기다리면 유실은 없지만 장애시간이 생긴다.

클러스터는 대부분 3대로 이뤄져 있어서 메시지 유실을 막을 수 있는데 프로듀서의 설정과 잘 써야 한다.

ack는 프로듀서가 발행한 데이터가 클러스터에 잘 저장되었는지 확인하는 것이다.

ack=0 확인 x, ack=1 리더에 저장확인, ack=all ISR에 저장되었는지 확인.

ISR의 최소 수는 설정값에 따라 다르다. 설정 값 2부터 의미 있다. 1은 리더 자신.

ISR설정은 복제 대수로와도 신경 써야 한다. 만약 3일 경우 브로커는 최소 3대 이상이어야 하며 한대라도 장애 나면 메시지를 받을 수 없다.

브로커가 3대 ack=all, replication, factor=3, min.insync.replicas=3은 하나 장애 시 메시지 받을 수 없음. 

프로듀서는 all설정으로 인해 리더와 파티션이 모두 복제되었는지 확인하고자 한다. 여기서 모든 리더와 파티션은 isr설정의 속한 리더와 파티션이다. 근데 isr설정이 3인데 factor가 3이니 복제는 3개의 파티션에 이뤄진다는 것을 알 수 있다. isr이 3의 의미는 모든 파티션이 데이터를 가지고 있어야 한다는 것인데 하나라도 장애가 난다면 아예 메시지를 받을 수 없는 상태가 되어버린다.

ISR그룹은 복제 시간에 따라 소속 파티션이 계속 변경될 수 있다. 카프카 시스템이 관리하는 개념 

ISR 그룹의 크기는 정해진 개수가 아님, min.insync.replicas설정은 데이터의 신뢰성과 안정성을 위해 사용자가 지정한 최소 복제본의 수를 기준으로 결정

min.insync.replicas는 거의 ack=all일 때만 의미가 있는 설정이다. 최소 복제 확인 수를 지정하는 것 이기에

 

프로듀서의 동작방식은 적어도 한번 전달. -> 1번 이상이기에 중복 가능, 설정을 통해 정확히 한번 가능

컨슈머를 운영하는 법

1. 여러 개의 컨슈머 프로세스 운영 2. 하나의 프로세스에 멀티 스레드 컨슈머 운영

가상환경이라면 후자가 비용 측면에서 좋음. 후자의 방식도 구현 방법이 여러 가지가 있음.

1. poll 하는 스레드 1개, 데이터 처리하는 스레드 N개 2. 컨슈머를 멀티 스레드로 운영

전자는 poll 하고 데이터 처리 오류 전에 auto커밋으로 커밋할 경우 데이터 유실.

둘 다 파티션 수만큼은 동작하도록 해야 함.

컨슈머 랙은 토픽의 최신 오프셋과 컨슈머 오프셋 간의 차이.

외부 솔루션으로 버로우가 있음. 버로우는 단순 랙이 증가했다고 컨슈머 에러로 보지 않음. 단시간에 트래픽이 몰린 것일 수 있으니. 랙은 있지만 컨슈머 오프셋이 따라잡으려고 하지만 잡지는 못할 경우 컨슈머 에러는 아님. 근데 컨슈머 오프셋이 멈춰있다면 컨슈머의 에러임.

 

컨슈머 배포전략은 중단과 무중단이 있음.

중단은 중단된 오프셋이 명확하기에 롤백에 유용함.

무중단은 실시간 서비스가 필요한 경우에 사용할 수 있음.

1. 블루그린: 구버전, 신규 버전 띄워놓고 트래픽 이전

2. 롤링 배포 : 인스턴스 2개 띄워놓고 하나씩 이전

3. 카나리 배포 : 극소수만 배포 후 안전하면 1,2 방법 중 하나로 배포.

위 3가지 모두 토픽의 데이터를 이어서 사용하려고 함. 리밸런싱은 필연적임. 1번은 파티션과 컨슈머의 수가 동일했다면 신규 띄우고 바로 구 버전 죽이면 되기에 리밸런싱이 한번 이뤄 짐. 2번은 인스턴스를 최소 2번 이상 죽이기 때문에 리밸런싱이 1번보다 많이 이뤄짐 

 

스프링 카프카에는 프로듀서는 카프카템플릿을 통해 데이터를 보낼 수 있다

스프링 카프카의 컨슈머는 2개의 타입으로 나뉨, 레코드리스너와 배치리스너인데 레코드별로 데이터를 가져오는지 여러 개를 가져오는지에 따라 나뉨.

스프링 카프카에서는 커밋의 방식을 ackMode라고 하는데 기본은 배치이다.-> poll로 가져온 레코드가 모두 처리된 이후에 커밋한다. 그니까 모든 비즈니스 로직이 끝난 후 커밋한다. 근데 만약 S3 같은 트랜잭션을 묶을 수 없는 곳에 적재 후 커밋 오류가 난다면 메시지 중복 처리가 될 것이다.

어노테이션 설정으로 컨슈머 스레드를 만들 수 있다. 파티션의 수만큼 만들면 성능이 올라간다.

다른 설정의 리스너나 리밸런스 리스너를 구축하기 위해선 커스텀 리스너 컨테이너를 써야 한다.

 

이벤트 수집보다 서비스 운영이 중요하다.

정책에 따라 데이터 파이프라인 운영 난이도가 달라진다. 중복이나 유실을 허용할 것인가..

멱등성 프로듀서는 정확한 한 번의 메시지 전달을 하지만(실제로는 여러 번 보낼 수 있지만 브로커에 적재가 pid값으로 인해 단 한 번만 되는 것) 컨슈머가 S3와 같은 트랜잭션에 묶을 수 없는 곳에 저장된 경우 단 한번만 적재되지는 않는다.

idempotence

정확한 단 한 번의 적재를 원한다면 UK를 지원하는 DB를 써라.

데이터 형식은 Json이 콘솔로 내용확인도 쉽고 프로듀서와 컨슈머의 스키마 변경으로 인한 재배포 비용이 낮아진다.

 

프로듀서 구축 시 고민거리

1, ack설정. all이면 데이터 안정성이 올라가지만 데이터 복제로 인해 느려질 수 있음

0과 1은 유실의 가능성이 다분함.

min-insync-replicas설정은 all일 때만 유효하기에 위 설정값에 좌지우지됨. ISR그룹에 따라 안정성을 보장하는 건데 0이나 1이면 복제의 의미가 없기 때문,

리트리가 설정은 중복 및 순서가 꼬일 수 있다. 압축은 네트워크 사용량은 줄지만 cup, 메모리 사용량이 올라간다.

 

토픽 구축 시 고민거리

1. 파티션 수는 메시지 순서를 지켜야 하냐에서 나눠진다. 파티션 한 개가 순서를 지키는데 수월하지만 리트라이나 여러 상황으로 꼬일 수 있다. 메시지 안에 발행 일시를 적으면 파티션이 여러 개라도 순서를 지킬 수 있는 조금의 대비책이 된다.

업데이트가 먼저 들어오면 로직 상에서 더 고민해 보자. 발행일을 기준으로 풀 수 있을 듯.

factor설정은 복제의 개수를 구하는 것인데 최대 브로커 대수만큼 할 수 있지만 느려질 수 있음. 그래도 브로커가 (리더파티션)이 죽었을 때 운영은 해야 하니까 2 이상으로 설정하는 것이 좋다. 1일 경우 복제를 하지 않는다는 것이고 복제가 없으면 리더가 죽었을 때 리더가 될 팔로워가 없으니 브로커는 데이터를 받지 않는다. 

factor설정과 isr설정과 ack설정의 연관성을 살펴보자면 factor설정으로 몇 개의 복제를 생성할 것 인지 설정하고 ack all로 프로듀서는 isr그룹의 팔로워들에게 모두 저장되었는지를 검사하고 isr 설정은 최소 몇개의 파티션에 복제가 되어야 하는지 설정하는 것이다. isr은 factor설정과 브로커 수보다 클 수 없다. 설정이 되는지 모르겠지만 논리상 크면 안 된다. 복제되었는지 검사를 하는데 복제 수 보다 많다면..

 

카프카는 타 시스템에 데이터 전달이 목적이고 eai는 타 시스템과의 연동을 해주는 컨버터 개념쯤 

 

컨슈머 멀티스레드일 경우 동일데이터 동시접근으로 인해 데드락에 유의하자. 파티션의 이름을 key로 사용하면 쉬운 해결책이 될 수 있다.

적재 시 파티션, 오프셋을 파일이름에 추가하면 쉽게 오류를 파악할 수 있다.

카프카 스트림즈는 수집된 지표데이터를 분기처리하고 필터링하는 동작에 좋다.

토픽을 복제해야 할 때는 미러링 2가 좋다.

 

카프카 복제 개념 글

 

[Kafka] Kafka Topic Replication

 

카프카 트랜잭션은 발행한 여러개의 이벤트를 하나의 트랜잭션으로 보기 위할 때 사용