프로듀서
프로듀서는 카프카 브로커로 데이터를 전송할때 내부적으로 파티셔녀, 배치 생성 단계를 거친다.
전송하고자 하는 데이터는 ProducerRecord 인스턴스를 생성하여 설정한다. 필수 파라미터는 토픽과 메시지 값이다.
ProducerRecord 생성시 추가 파라미터를 사용하여 오버로딩하여 ProductRecord의 내부 변수를 선언할 수 있다.
파티션 번호를 직접 지정하거나 타임스탬프를 설정, 메시지 키를 설정할 수도 있다.
KafkaProducer 인스턴스가 send()를 호출하면 ProducerRecord는 파티셔녀(partitioner)에서 토픽의 어느 파티션으로 전송될 것인지 정해진다.
Properties configs = new Properties();
configs.put...
KafkaProducer<String, String> producer = new KafkaProducer<>(configs);
producer.send(record)
KafkaProducer 인스턴스를 생성할때 파티셔너를 따로 설정하지 않으면 기본값인 DefaultPartitioner로 설정되어 파티션이 정해진다.
파티셔너에 의해 구분된 레코드는 데이터를 전송하기 전에 어큐뮬레이터(accumulator)에 데이터를 버퍼로 쌓아놓고 발송한다.
버퍼로 쌓인 데이터는 배치로 묶어서 전송함으로써 카프카의 프로듀서 처리량을 향상시키는데 상당한 도움을 준다.
파티셔너 제공
프로듀서 API를 사용하면 UniformStickyPartitioner, RoundRobinPartitioner 2개의 파티션을 제공한다.
카프카 클라이언트 라이브러리 2.5.0 버전에서 파티셔너를 지정하지 않은 경우 UniformStickyPartitioner가 파티셔너로 기본 설정된다.
공통점
메시지 키가 있을 떄는 메시지 키의 해쉬값과 파티션을 매칭하여 데이터를 전송한다.
차이점
UniformStickyPartitioner이 RoundRobinPartitioner의 단점을 개선했다.
UniformStickyPartitioner은 메시지 키가 없을때 파티션에 최대한 동일하게 분배하는 로직이 들어있다.
UniformStickyPartitioner는 프로듀서 동작에 특화되어 높은 처리량과 낮은 리소스 사용률을 가진다.
RoundRobinPartitioner은 ProducerRecord가 들어오는 대로 파티션을 순회하면서 전송하기 때문에 배치로 묶이는 빈도가 적다.
될 수 있으면 많은 데이터가 배치로 묶여 전송되어야 성능 향상이 되므로 UniformStickyPartitioner가 기본 파티셔너로 설정되었다.
UniformStickyPartitioner는 어큐뮬레이터에서 데이터가 배치로 모두 묶일때까지 기다렸다가 배치로 묶인 데이터는 모두 동일한 파티션에 전송함으로써 성능이 더 향상되었다.
Partitioner 인터페이스 제공
카프카 클라이언트 라이브러리에서는 사용자 지정 파티셔너를 생성하기 위한 Partitioner 인터페이스를 제공한다.
Partitioner 인터페이스를 상속받은 사용자 정의 클래스에서 메시지 키 또는 메시지 값에 따른 파티션 지정 로직을 적용할 수도 있다.
압축 방식 지정 가능
카프카 프로듀서는 압축 옵션을 통해 브로커로 전송시 압축 방식을 정할 수 있다.
압축 옵션을 정하지 않으면 압축이 되지 않은 상태로 전송된다.
카프카 프로듀서에서는 압축 옵션으로 gzip, snappy,. lz4, zstd를 지원한다.
압축을 하는데에 CPU 또는 메모리 리소스를 사용하므로 사용 환경에 따라 적절한 압축 옵션을 사용하는 것이 중요하다.
프로듀서 주요 옵션
필수 옵션과 선택 옵션이 있는데, 선택 옵션을 설정하지 않으면 선택 옵션의 기본값(default)을 파악해서 운영해야한다.
필수 옵션
- bootstrap.servers
프로듀서가 데이터를 전송할 대상 카프카 클러스터에 속한 브로커의 hostname:port 를 1개 이상 작성한다.
2개 이상 브로커 정보를 입력하여 일부 브로커에 이슈가 발생하더라도 접속하는데에 이슈가 없도록 설정 가능하다.
- key.serializer
레코드의 메시지 키를 직렬화하는 클래스를 지정한다.
- value.serializer
레코드의 메시지 값을 직렬화화는 클래스를 지정한다.
선택 옵션
- acks
프로듀서가 전송한 데이터가 브로커들에 정상적으로 저장되었는지 전송 성공 여부를 확인하는데에 사용하는 옵션이다.
| value | 설명 |
| 0 default | 프로듀서가 전송한 즉시 브로커에 데이터 저장 여부와 상관 없이 성공으로 판단한다. |
| 1 | 리더 파티션에 데이터가 저장되면 전송 성공으로 판단한다. |
| -1 or all | min.insync.replicas 개수에 해당하는 리더 파티션과 팔로워 파티션에 데이터가 저장되면 성공하는 것으로 판단한다. |
- buffer.memory
브로커로 전송할 데이터를 배치로 모으기 위해 설정할 버퍼 메모리양을 지정한다. 기본 값은 32MB이다.
- retries
프로듀서가 브로커로부터 에러를 받고 난 뒤 재전송을 시도하는 횟수를 지정한다. 기본값은 2147483647이다.
- batch.size
배치로 전송할 레코드 최대 용량을 지정한다. 너무 작게 설정하면 프로듀서가 브로커로 더 자주 보내기 때문에 네트워크 부담이 있고 너무 크게 설정하면 메모리를 더 많이 사용하게 되는 점을 주의해야한다. 기본값은 16384이다.
- linger.ms
배치를 전송하기 전까지 기다리는 최소 시간이다. 기본값은 0이다.
- partitioner.class
레코드를 파티션에 전송할때 적용하는 파티셔너 클래스를 지정한다.
기본값은 org.apache.kafka.clients.producer.internals.DefaultPartitioner이다.
- enable.idempotence
멱등성 프로듀서로 동작할지 여부를 설정한다. 기본값은 false다.
- transactional.id
프로듀서가 레코드를 전송할때 레코드를 트랜잭션 단위로 묶을지 여부를 설정한다.
프로듀서의 고유한 트랜잭션 아이디를 설정할 수 있다.
이 값을 설정하면 트랜잭션 프로듀서를 동작한다. 기본값은 null이다.
'Coding > Apache Kafka' 카테고리의 다른 글
| [아파치 카프카 어플리케이션 프로그래밍] 7. 프로젝트 생성하여 카프카 컨슈머 실행 (2) | 2022.05.21 |
|---|---|
| [아파치 카프카 어플리케이션 프로그래밍] 6. 특정 파티션 설정, 프로듀서 전송 결과 Callback 클래스 생성하기 (0) | 2022.05.20 |
| [아파치 카프카 어플리케이션 프로그래밍] 4. 프로젝트 생성하여 카프카 프로듀서 실행 (0) | 2022.05.19 |
| [아파치 카프카 어플리케이션 프로그래밍] 3. 카프카의 기본 개념 정리 (브로커, 페이지 캐시, 리플리케이션, 컨슈머 그룹, 코디네이터, 주키퍼, 토픽, 레코드, 오프셋) (0) | 2022.05.18 |
| [아파치 카프카 어플리케이션 프로그래밍] 2. 카프카 프로듀서, 컨슈머 실행해보기 (with Docker) (0) | 2022.05.18 |