Kafka 튜닝, 최적화 방안

**이 글은 kafka3 이전 버전 기준으로 쓰여졌음을 알립니다

 

성능목표 이해

Kakfa는 사용하려는 서비스 목적에 따라 성능 목표를 크게 Throughput, Latency, Durability, Availability 네가지로 분류할 수 있습니다. 각 목표는 상호 trade-off 관계로 모든 목표를 동시에 모두 최적화할 수 없다.

 

·         Thoughput: kafka 특성상 많은 데이터를 빠르게 쓰는 것은 문제가 없음

·         Latency: 하나의 메시지를 가능한 빠르게 전달 (producer -> broker -> consumer)

·         Durability: 메시지의 유실을 최소화, 이벤트 기반 마이크로서비스 또는 데이터 수집 파이프라인

·         Availability: kafka 서버의 다운타임 최소화, 장애 시 가장 빠르게 복구해야 하는 서비스

 

보통

throughput <-> latency

durability <-> availiability

이러한 trade-off 관계를 가진다.

  

따라서 서비스의 특징에 따라 성능목표를 정한 후 파라미터 튜닝이 필요하다.

어떤 서비스 목표를 최적화할지 결정 ->카프카 클러스터와 클라이언트 설정 -> 벤치마크, 모니터링, 튜닝

---

 

카프카 모니터링에 필요한 metric이해

metric은 타임스탬프와 보통 한두가지 숫자 값을 포함하는 이벤트를 의미하고, 로그와 달리 주기적으로 발생한다. 보통 로그는 무언가 발생했을 때 로그파일에 기록되는 반면, metric은 주기적으로 발생하는 또는 수집하는 이벤트라 볼 수 있다.

 

1. Throughput (단위 시간당 전송 데이터 량) 최대화
   •  producer
     • 파티션 수 증가: 병렬적으로 처리
     • Batch size 증가: 한번에 보내는 양(레코드를 묶는 양)을 증가시켜 전송 횟수 감소
     • Linger.ms 조정: linger.ms: batch형태의 메시지를 보내기 전에 추가적인 메시지들을 위해 기다리는 시간을 조정할 수 있음. 0보다 큰 값을 설정하면 지연은 발생하지만 처리량은 좋아짐 (default:0, 지연 없음), 또는 broker가 전송 전에 producer가 추가 메시지를 기다리는 시간. 보통 producer에서 Batch.size가 가득 차서 전송이 되거나, linger.ms 시간이 다 돼서 전송이 됨. Batch.size에 도달하면 linger.ms 값과 상관없이 보내지만, 도달하지 않아도 설정된 시간이 흐르면 전송을 함.
     • Compression.type : 메시지 압축 타입, LZ4, gzip, snappy 등이 있다.
     • Buffer.memory(레코드를 축적하는 역할) : producer가 보내지 못한 메시지를 보관하는 메모리 크기 조정, 메모리가 full이 되면, 메시지 전송을 막음, 메모리에 여유가 생기거나, max.block.ms를 초과하면 전송 // max.block.ms로 명시된 시간이 지나도 메모리가 확보되지 않으면 TimeOutException이 발생.파티션이 많지 않으면, 조정할 필요 없음, 파티션이 많다면, 메모리를 늘려서 블락없이 더 많은 데이터가 전송되도록 설정
   •  consumer
     • consumer : partition = 1 : 1 비율로 설정
     • fetch.min.bytes(consumer가 가져오는 최소 데이터 사이즈, 기본값 = 1byte): 이 값보다 적은 데이터가 있으면 브로커에서 보내지 않음, 값을 증가시키면 producer의 batch.size 증가와 동일 효과
     • fetch.max.wait.ms: consumer에서 데이터를 가져오는 최소 시간 (기본값 = 500ms): 새로운 데이터가 입력되어도, 해당 시간 이전에는 가져가지 않음, 내부적으로 consumer가 fetch 요청을 해도, 브로커가 보내지 않음.

       ---

2. Latency 최소화 방안
   • broker
     • 파티션 개수 제한: 너무 많은 파티션은 메시지 지연을 유발, 파티션 복사를 위한 시간만큼 지연
     • Broker수 증가, 파티션 수는 적게 설정: 하나의 브로커에서 감당하는 파티션 수를 감소시켜 복제에 소요되는 시간을 최소화,
     • Num.replica.fetchers(기본값=1): Leader에서 메시지를 복제할 때 사용할 broker 내부의 fetcher thread 수로, thread수를 늘리면 I/O 병렬 비율과 fetcher throughput이 높아짐, follower broker 의 I/O 병렬 수준을 정의, Leader broker에서 데이터를 복제하는 thread의 개수
   • producer
     • Linger.ms: producer가 전송 전에 대기하는 시간(기본값 0= 데이터를 수집하는 순간 broker로 전송 -> 지연 시간, 처리량 감소): batch.size가 꽉 찰 수 있도록 시간 설정,
     • Compression.type: 압축이 필요한지 검토
       • CPU: 압축을 위한 자원 사용
       • NW(Network): 압축된 경우 NW bandwidth 사용량 줄어듦.//bandwidth(대역폭: 물리적으로 한번에 보낼 수 있는 최대 데이터 량)
       • 압축 성능에 따라 CPU 사용, NW bandwidth 줄여서 지연 최소화 가능
     • Acks: Acknowledgments로 producer가 메시지를 보내고, 그 메시지를 kafka가 잘 받았는지 확인할지 말지 결정하는 옵션. 언제 broker로부터 메시지 전송 결과를 받을 것인지 정의
       • Acks = 1: 데이터 복제 없이, 원본만 저장되면 결과 리턴
       • Acks = 0: 받았는지 확인하지 않고 계속 전송
       • Acks = all: 데이터 복제까지 설정한 값만큼 성공적으로 된 것을 확인하고 결과 리턴
   • consumer
     • fetch.min.bytes(기본값 = 1): broker에서 데이터를 가져오는 최소 size,
       • fetch.min.bytes= 1: 1byte만 있어도 요청 시 바로 전송 (지연 없음)

         ---

3. Durability 보장 방안(메시지 유실 최소화)
   • broker
     • replication.factor
       • 3: 높은 수준의 durability 지원
     • default.replication.factor: auto.create.topics.enable 가 true 인 경우(producer에 의해서 메시지를 broker에 전송했는데, 존재하지 않는 topic에 메시지를 전송한 것이라면 해당 topic을 자동으로 생성하는 것), 자동으로 생성되는 topic의 복제 수 설정 -> 높을수록 높은 durability 보장
     • broker.rack: 하나의 rack(서버 rack과 비슷한 느낌)에 broker가 동작하지 않도록 설정
       • rack 파워가 나가면 그 서버가 다 나가므로, 1개의 rack에 다수의 broker를 몰아 넣는 것은 위험, 다수의 rack 에 분산하여 브로커 옵션(broker.rack) 설정 및 배치(deploy)
       • 안정성은 높아지지만 데이터 복제 시 NetWork 부하 증가
     • unclean.leader.election.enable: broker가 죽었을 때, OSR replica도 leader로 선택될 수 있도록 설정 (true)
       • OSR(Out-of-sync):  Topic 별로 메시지 원본을 저장하는 broker(leader)가 선정되고, 이를 복제할 broker(follower)가 결정된다. 이때 leader에 저장된 메시지와 follower에 저장된 데이터간의 차이(복제 시간)가 커지면 OSR이라는 상태가 나타나게 된다.  죽은 broker의 최신 메시지를 받지 못한 replica, 즉 데이터 유실 가능
       •  유실보다 서비스 가용성을 높이는 경우 -> unclean.leader.election.enable = true
       • 유실을 최소화하는 경우 -> unclean.leader.election.enable = false
     • log.flush.interval.ms : 메시지가 디스크로 플러시 되기 전에 모든 주제의 메시지가 메모리에 보관되는 최대 시간, 설정되지 않은 경우 log.flush.schedular.interval.ms값이 사용된다. 최소값은 0.
       • log.flush.interval.messages : 메시지가 디스크로 플러시 되기 전 로그 파티션에  누적된 메시지 수 à 입력된 메시지를 메모리 (페이지 캐시)에서 디스크로 저장하는 수준
         • 값이 클수록 disk I/O 적게 발생 → 메모리 데이터 유실 가능
         • 값이 작을수록 disk i/o 많이 발생 → 메모리 데이터 유실 거의 없음
   • producer
     • Acks = all (acks = -1과 동일): producer는 자신이 보낸 메시지에 대해 kafka의 leader와 follower까지 받았는지 기다림. 최소 하나의 복제본까지 처리된 것을 확인하므로 메시지가 손실될 확률은 거의 없음.
       • 모든 replica에 복제가 완료된 후, producer에 ack 리턴
       • min.insync.replicas: acks =all 일 때 broker가 ack 를 보내기 위한 최소 복제본의 수(만약  min.insync.replicas = 2이면, 적어도 2개의 follower에게 성공적으로 replication이 되면 ack를 producer에게 보냄)
     • Retries: 전송 실패 시 자동으로 재전송하는 횟수
       • 일시적 오류로 producer가 2번 전송할 가능성 있음 또는메세지 순서 변경될 가능성 있음.
       • 한번에 여러 번의 request가 network에 대기중인 경우, fail된 request 다음 메시지가 먼저 전송되는 경우 발생
         • Max.in.flight.requests.per.connection = 1로 설정 (한번에 1개 요청) //"max.in.flight.requests.per.connection"이 2 이상인 경우 일시적인 에러로 인한 재시도 때문에 메시지 전송 순서가 바뀔 수 있다. 예를 들어 producer가 "메시지1", "메시지 2" 순서로 전송했을때, max.in.flight.requests.per.connection 값이 1이면 한 번에 한 개의 메시지만 보낸다. 따라서 "메시지 1"이 정상적으로 전송되고 나서 "메시지 2"를 보낸다. max.in.flight.requests.per.connection 값이 2면 한 번에 두 개의 메시지를 보내게 된다. "메시지 1"과 "메시지 2"를 전송했는데 타이밍 이슈로 "메시지 2"는 전송에 성공하고 "메시지 1"만 실패한 경우, producer는 "메시지 1"을 재전송하게 된다. 결국 브로커는 "메시지 2", "메시지 1" 순으로 메시지를 받게 된다.
   • Consumer
     
     • enable.auto.commit/ auto.commit.interval.ms
       • poll()을 호출할 때마다 commit 할 시간이 되었는지 확인하고, poll 요청으로 가져온 마지막 offset을 commit
       • Offset 정보가 commit 되지 않은 상태에서 장애 발생 후, 복구 시, 메시지를 중복 소비할 가능성이 있음.
       • auto.commit.enable = false 사용, manual commit(commitSync(),commitAsync())을 사용하면 해결 가능.

         ---

4. Availability 보장 방안(장애 복구를 빠르게)
   • Broker
     • 너무 많은 파티션 수 제한
       • 파티션 별 리더 선출에 많은 시간 소요
     • Min.insync.replicas: producer가 응답을 받기 위한 최소 복제 수
       • 값이 크면, 복제 실패 시 producer 장애 유발 -> durability 높음
       • 값이 1 이면, 원본만 저장되면 producer 정상 동작 -> durability 낮음
     • Unclean.leader.election.enable
       • broker 가 죽었을 때, OSR replica도 leader로 선택될 수 있도록 설정 (true)
     • Num.recovery.threads.per.data.dir: broker가 시작/ 중지할 때, 다른 broker와 sync를 맞추기 위해 log data file을 스캔하는 thread 개수
       • 각 로그 디렉터리를 기준으로 복구 할 때 사용되는 thread의 개수를 말한다. 브로커의 시작과 종료 시에만 사용되므로 병행 처리를 하도록 많은 수의 thread를 지정하는 것이 좋다. logs.dirs가 3개고 이 값이 5라면 15개의 thread가 실행된다.
       • 값이 크면 한번에 여러 log 파일을 동시에 처리 가능 (RAID 구성인 경우) -> 즉 broker의 구동 속도가 빠름
   •  Consumer
     • session.timeout.ms: consumer가 비정상적으로 종료되었을 경우, broker가 장애로 인지하는 최소 시간 
       • consumer와 broker사이의 session timeout 시간, 만약 consumer가 그룹 코디네이터에게 하트 비트를 보내지 않고 session.timeout.ms 시간이 지나면 해당 consumer는 종료되거나 장애가 발생하는 것으로 판단하고 consumer 그룹은 rebalance 시도
       • session.timeout.ms를 기본값보다 낮게 설정하면 실패를 빨리 감지하지만(복구가 빠름) 작업을 완료하는 시간이 길어지게 되면(조금만 지연되도, failure로 판단) 원하지 않게 rebalance가 일어날 수 있음
       • 반대로 session.timeout.ms 를 기본값보다 높게 설정하면 오류 감지 하는데 오랜 시간이 걸리지만, rebalance 발생 가능성이 줄어듦

 

 

 

Producer 및 Consumer 수에 따른 결론

하드웨어 스펙이 좋으면 Producer, Consumer 수가 증가할수록 처리량은 늘어난다.

 

•  Consumer와 partition의 이상적인 비율
  • Kafka 에서 consumer group의 consumer의 개수는 partition의 개수와 되도록 일치시키는 것이 효율적이다. Partition이 consumer보다 많으면, 하나의 consumer가 여러 개의 partition을 처리해야 하기 때문에 지연이 생길 수 있다.
  • 반대로 consumer의 수가 partition의 수보다 많다면, 놀고 있는 consumer가 생기기 때문에 비용을 낭비하게 된다.

 

 

 

참고자료

https://www.slideshare.net/freepsw/apache-kafka-metrics-123663954

https://m.blog.naver.com/PostView.naver?isHttpsRedirect=true&blogId=freepsw&logNo=221073557983

https://server-engineer.tistory.com/846 [HelloWorld]

https://devidea.tistory.com/90

https://ohjongsung.io/2020/01/04/%EC%B9%B4%ED%94%84%EC%B9%B4-%ED%8A%9C%EB%8B%9D-%EB%B0%A9%EC%95%88-%EC%A0%95%EB%A6%AC