[Kafka] 2. Kafka Producer

 

 

1. What is a Kafka Producer?

카프카 프로듀서는 토픽에 메시지를 보내는 주체이다.

이때 메시지는 다음과 같은 내용들을 포함한다.

-메시지는 어떤 데이터든 될 수 있고, 바이트배열로 직렬화 되어야 한다.

-메시지의 키에 따라 토픽에 어느 파티션에 쓰여질 지 정해진다.

-프로듀서는 메시지의 타임스탬프를 적는다.(선택적)

-압축 타입은 메시지의 크기를 줄인다.(선택적)

-메시지의 헤더에는 추가적인 메타데이터가 포함된다.(선택적)

-브로커는 메시지를 수신하면 파티션과 offset ID를 추가한다.

 

프로듀서는 적절한 구조로 메시지를 만들고 메시지를 카프카 토픽에 보내는 역할을 한다.

 

1.1. Connecting your producer to Kafka

프로듀서는 일단 카프카 bootstrap server(카프카 브로커들의 일부)에 연결해서 카프카 브로커 서버들의 주소들을 찾는다.

그리고 각 토픽의 파티션의 리더가 어느 브로커인지 찾는다.

이러한 과정은 프로듀서가 첫 번째 브로커에게 MetaDataRequest를 보냄으로써 처리된다.

따라서 메시지에는 메시지 내용뿐만 아니라 타겟 토픽, 파티션에 대한 정보도 포함되어 있다.

 

그 다음 프로듀서는 해당 메시지를 독점적인 TCP프로토콜을 이용해서 리더 브로커에게 보낸다.

이는 파티션 내에서 메시지 순서를 유지하면서 확장될 수 있게 해준다.

 

이러한 과정은 프로그래밍 언어에서 프로듀서가 생성될 때 백단에서 처리된다.

from kafka import KafkaProducer
producer = KafkaProducer(bootstrap_servers='localhost:9092')

 

 

 

2. Message Serialization in Kafka Producer

카프카 메시지 직렬화를 위해 적절한 직렬화 함수를 사용해야 한다.

producer = KafkaProducer(value_serializer=lambda v: JSON.dumps(v).encode(' utf-8'))

 

위 코드에서 프로듀서의 메시지는 카프카 리더 브로커에 보내지기 전에 Json에서 byte배열 형태로 직렬화된다.

그리고나서 아래와 같이 보낼 수 있다.

producer.send('test-topic', key='key1', value='{"value": "value1"}')

 

 

3. Message Keys and Partitioning

카프카는 여러 파티션에 메시지를 쓰면서 확장성과 성능을 향상시킨다.

이 과정은 가끔 다른 브로커 서버에 있는 리더와 통신을 포함한다.

그러나 이러한 확장성은 메시지 순서를 희생하는 돼서 온다.

 

메시지들은 오로지 같은 파티션 내에서만 순서가 보장된다.

그래서 각 메시지는 메시지의 순서를 나타내는 적절한 키를 갖는 것이 중요하다.

메시지의 키 값으로는 예를 들면 account number, IOT device Id등이 올 수 있다.

 

프로듀서의 파티셔너는 메시지 키를 해싱하여 올바른 파티션에 메시지가 전송되도록 결정하고, 지정된 파티션에 매핑한다.

만약 메시지 키를 따로 설정하지 않았다면 라운로빈 방식이나 스티키 파티셔닝 전략에 따라 메시지가 보내진다.

라운드로빈 방식은 단순히 파티션의 순서대로 메시지가 보내진다.

스티키 파티셔닝은 시간이나 배치 사이즈가 트리거 되기 전까지 하나의 파티션에만 메시지를 쌓는 방식이다.

 

아래 명령으로 토픽에 구체적인 설정들을 확인할 수 있다.

docker exec -it kafka kafka-topics --describe --topic test-topic --bootstrap-server localhost:9092

 

 

4. Durability and Error Handling

카프카는 프로듀서에 여러가지 설정을 제공하며 이를 통해 내구성과 성능간의 밸런스를 조절할 수 있다.

 

4.1. Acks

acks설정은 프로듀서가 메시지를 보내기 전에 몇 개의 브로커가 메시지 수신을 확인할 것인지에 관한 설정이다.

acks=all 설정은 모든 복제 파티션들이 메시지 수신을 완료해야지 메시지 전송이 완료된다.

이 숫자는 acks=0으로 줄일 수 있다. 이렇게 하면 메시지가 다른 파티션에 복제되기 전에 다른 메시지가 전송된다.

 

아래와 같이 acks를 설정할 수 있다.

producer = KafkaProducer(bootstrap_servers='localhost:19092, acks='all')

 

 

프로듀서가 아닌 브로커쪽에서도 관련된 설정이 있다.

* replication.factor : 토픽에 쌓이는 메시지를 몇 개 복제할 것인지

* min.insync.replicas : acks=all이 설정되었을 때 필요한 최소 복제본 개수

 

아래와 같이 토픽 파라미터를 설정할 수 있다.

docker exec -it kafka kafka-configs --bootstrap-server localhost:9092 --entity-type topics --entity-name your-topic-name --alter --add-config min.insync.replicas=2

 

 

4.2. Delivery Settings

프로듀서쪽에서 전달관련된 설정들이다.

* retries : 메시지 전달 실패시 몇 번 재시도할 것인지

* retry.backoff.ms : 메시지 재전달하기 전에 얼마나 기다릴 것인지

* delivery.timeout.ms : 프로듀서가 메시지 보내고 확인 응답을 받을 때까지 최대로 얼마나 기다릴 것인지

 

이러한 설정은 프로듀서를 생성할 때 설정할 수 있다.

 

 

5. Testing Your Kafka Producer

Kafka 보안은 전송 중 데이터에 대한 SSL/TLS 암호화, 인증을 위한 SASL, 주제에 대한 권한 있는 접근을 위한 ACL을 포함합니다. 

프로듀서는 브로커의 암호화 설정에 맞는 관련 구성을 필요로 하며, 해당 주제에 접근하기 위해 적절한 자격 증명을 제공해야 합니다.

Kafka 프로듀서의 성능은 처리량, 지연 시간 및 리소스 활용 측면에서 측정할 수 있습니다. 이를 제어하는 가장 중요한 설정은 batch.size와 linger.ms 설정입니다.

* Batch Size : 배치 크기(Batch Size)는 파티션으로 전송되는 메시지 배치의 최대 크기(바이트 단위)입니다. 배치 크기를 늘리면 처리량이 향상될 수 있지만 프로듀서의 메모리 요구 사항과 지연 시간에 부정적인 영향을 미칠 수 있습니다.
* Linger Time : 린거 시간(Linger Time)은 메시지를 브로커에 배치로 전송하기 전에 데이터를 버퍼링하는 최대 시간(밀리초)입니다. 린거 시간이 길어지면 배치 크기가 커지고 처리량이 증가할 수 있지만, 그 대가로 지연 시간이 증가할 수 있습니다.

Kafka는 또한 프로듀서의 성능과 상태에 대한 통찰력을 제공하는 내장 메트릭을 제공하며, 여기에는 처리량, 지연 시간, 버퍼 가용성 및 오류율이 포함됩니다.

print(producer.metrics())

 

벤치마킹, 병목 현상 식별, 그리고 반복적인 배치 및 린거 설정 조정을 통해 프로듀서를 사용 사례의 요구 사항에 맞게 최적화할 수 있습니다.