카프카 CLI 활용

[manjeong-dev]

선택 옵션이 없다면 브로커에 설정된 기본 설정값 또는 커맨드 라인 툴의 기본값으로 대체되어 설정됨

• 브로커에 설정된 파티션 기본값으로 지정되므로.. 파티션 개수 줄일 수는 없음!!!!
  • 삭제 대상 파티션을 지정 & 기존에 저장되어 있던 레코드를 분산하여 저장 필요
  • 피치못하게 파티션 개수를 줄여야 할 때는 토픽을 새로 만드는 편이 나음

주키퍼를 실행한 후 카프카 브로커를 실행해야 함!!

bin/zookeeper-server-start.sh config/zookeeper.properties : 주키퍼 실행 bin/kafka-server-start.sh config/server.properties : 카프카 실행 bin/kafka-broker-api-versions.sh --bootstrap-server localhost:9092 : 카프카 설정값들 확인

kafka-topics.sh bin/kafka-topics.sh --bootstrap-server localhost:9092 --list : 카프카 브로커의 토픽에 있는 리스트 확인 bin/kafka-topics.sh --create --bootstrap-server my-kafka:9092 --topic hello.kafka : 카프카 토픽 생성 bin/kafka-topics.sh --bootstrap-server localhost:9092 --topic hello.kafka --describe : 토픽 설정값 확인

bin/kafka-topics.sh --create --bootstrap-server my-kafka:9092 --partitions 10 --replication-factor 1 --topic hello.kafka2 --config retention.ms=172800000 : 설정값 넣어서 토픽 생성

• retention : 레코드 유지 기간

Topic: hello.kafka2 PartitionCount: 3 ReplicationFactor: 1 Configs: segment.bytes=1073741824,retention.ms=172800000 Topic: hello.kafka2 Partition: 0 Leader: 0 Replicas: 0 Isr: 0 Topic: hello.kafka2 Partition: 1 Leader: 0 Replicas: 0 Isr: 0 Topic: hello.kafka2 Partition: 2 Leader: 0 Replicas: 0 Isr: 0 leader : 리더 브로커는 몇번에? (broker-0)

bin/kafka-topics.sh --bootstrap-server localhost:9092 --topic test --alter --partitions 10 : alter로 설정값 변경

kafka-configs.sh --add-config min.insync.replicas : 프로듀서, 컨슈머 데이터 통신시 얼마만큼의 양으로 보낼지, 워터마크 용도 bin/kafka-configs.sh --bootstrap-server my-kafka:9092 --broker 0 --all --describe

kafka-console-producer.sh 파티션에 레코드를 넣을 때 value 만 넣거나 key.separator를 이용해 k:v 로 넣을 수도 있음 메시지 키가 null 일 경우가 존재할 수 있음

• 메시지키 : 데이터 구분자 용도
• 특정 레코드에 대해 순서를 지키고 싶다, 같이 처리되고 싶다 싶을 때 사용
• 같은 메시지키는 같은 파티션에만 들어감(들어간 순서를 지킬 수 있음)
• 메시지키가 없으면 파티션 중에 Round-robin으로 들어감 bin/kafka-console-producer.sh --bootstrap-server my-kafka:9092 --topic hello.kafka : 토픽에 데이터 넣기 bin/kafka-console-producer.sh --bootstrap-server my-kafka:9092 --topic hello.kafka --property "parse.key=true" --property "key.separator=:"

kafka-console-consumer.sh 토픽에 있는 데이터를 조회 --from-beginning : 토픽에 저장된 가장 처음 데이터부터 출력 --max-messages : 최대 컨슘 메시지 개수 설정 --partition : 특정 파티션만 컨슘 (메시지 키에 해당하는 파티션만 확인해보고 싶을 때도 사용할 수 있음) --group : 컨슈머 그룹 기반으로 동작 컨슈머 그룹

• 특정 offset까지 읽었다는 것을 commit 시키기 위한 용도
• 컨슈머 그룹으로 토픽의 레코드를 가져갈 경우 어느 레코드까지 읽었는지에 대한 데이터가 카프카 브로커에 저장됨

bin/kafka-console-consumer.sh --bootstrap-server my-kafka:9092 --topic hello.kafka --from-beginning --property print.key=true --property key.separator="-"

• __consumer_offsets 라는 토픽에 consumer group 커밋 내용이 저장됨

kafka-consumer-groups.sh

• 따로 생성 명령어 x, 컨슈머 동작시 컨슈머 그룹이름 지정하면 생성됨
• 해당 컨슈머 그룹이 어떤 토픽을 대상으로 몇번 오프셋까지 가져갔는지. 상태. 파티션 번호. 현재까지 가져간 레코드의 오프셋. 파티션 마지막 레코드의 오프셋. 컨슈머 LAG. 컨슈머 ID, 호스트
• 컨슈머 LAG : 마지막 레코드의 오프셋 - 현재까지 가져간 레코드의 오프셋
  • 지연의 정도를 확인(컨슈머와 프로듀서 처리량 차이를 확인할 수 있음)

bin/kafka-console-consumer.sh --bootstrap-server my-kafka:9092 --topic hello.kafka --group hello-group : 그룹 지정 컨슘 bin/kafka-consumer-groups.sh --bootstrap-server my-kafka:9092 --group hello-group --describe

오프셋 리셋

bin/kafka-consumer-groups.sh --bootstrap-server my-kafka:9092 --group hello-group --topic hello.kafka --reset-offsets --to-earliest --execute (모든 데이터 재처리 용도)

퍼포먼스 측정 kafka-producer-perf-test.sh kafka-consumer-perf-test.sh

• 카프카 브로커와 컨슈머(해당 스크립트를 돌리는 호스트)간의 네트워크 체크

kafka-reassign-partitions.sh

• 한 브로커가 핫스팟이 되서 데이터가 너무 몰릴 때(리더파티션이 한 브로커에 너무 몰릴 때 팔로워 파티션을 가진 브로커와 리더파티션 위치 변경을 통해 재조정)
• auto.leader.rebalance.enable
  • 기본값 true. 자동으로 클러스터 단위에서 리더 파티션을 rebalancing
  • 브로커의 백그라운드 스레드가 일정한 간격으로 리더 위치 파악 > 리더 rebalancing

kafka-delete-record.sh

• offset 으로 지정된 곳까지 삭제 (0~5)

kafka-dump-log.sh

• segment log 볼 때

토픽 생성 방법

• --create 로 CLI로 명시적으로 생성
• 프로듀서나 컨슈머가 카프카 브로커에 생성되지 않은 토픽에 대해 데이터를 요청할 때
  • broker 옵션 기준으로 생성됨
  • 브로커 실행시 auto create 옵션 : true > 신규 생성 자동으로

카프카 브로커 <> CLI 툴 버전을 맞추자!

[manjeong-dev]

그룹 없으면 항상 첫번째부터 읽는지? --from-beginning 같은 옵션없으면 안읽어짐

• ㅇㅇ

[manjeong-dev]

https://effectivesquid.tistory.com/entry/Anatomy-Kafka-Consumer1

하나의 파티션은 하나의 컨슈머에서만 메시지 소비 가능 하나의 컨슈머는 여러개의 파티션 메시지를 소비할 수 있음

rebalancing 이 자주 발생하면 모든 consumer가 메시지 소비를 중단. 자주 일어나면 안됨!

한 파티션 내 읽는 순서는 보장 / 다른 파티션과의 읽는 순서는 보장되지 않음 (한 파티션 내) but 처리 순서는 보장 되지 않음 why? 처리는 컨슈머가 하는거고, 읽고 나서 컨슈머가 병렬처리하면 처리 순서는 바뀔 수 있음

[manjeong-dev]

acks all option은 replication 이 잘됐는지 ack 받는 replication factor가 커도 성능상 이슈가 안될 수 있음(ack 받는 옵션을 작게 두면)

[manjeong-dev]

Q. 커맨드 라인 툴을 통해 토픽 관련 명령을 실행할 때 선택 옵션을 지정하지 않으면 브로커에 설정된 기본 설정값으로 설정되지 않는다 => X Q. 카프카 토픽이 사용하는 파티션의 수를 줄일 경우에는 새로운 토픽을 만드는 것이 낫다 => O Q. kafka 서버를 띄울 때 broker -> zookeeper 순으로 띄워야 한다. => X Q. Kafka에서 토픽을 생성할때 필수 옵션인것을 모두 고르시오. 브로커 주소 주키퍼 주소 토픽 파티션 개수 토픽 이름 => 1,4 Q. kafka CLI로 할 수 없는 작업을 모두 고르시오? 토픽 목록 조회 토픽 설정 수정 브로커 설정 정보 조회 브로커 개수 조정 토픽 파티션 개수 조정 => 4 Q. (같은 토픽 내) key 가 null 인 레코드들은 동일한 key 를 가지므로, 동일한 파티션으로 전송된다. => X (round-robin 2.5>sticky(n개씩 묶어 넣는다)) Q. 레코드에서 key:value 형식으로 지정하는 명령어는 --property “parse.key=true” --property “key.separator=” 이다. 레코드에 key를 지정하는 것의 장점 중 아닌 것은? 레코드에 대한 컨슈머가 읽는 순서 정의 레코드에 대한 컨슈머가 처리하는 순서 정의 레코드을 일정 그룹으로 만들어 처리할 수 있도록 함 => 2 Q. 컨슈머의 처리량을 가늠할 수 있는 consumer-groups 정보 중 하나는 OOO 이다. => LAG Q. kafka-consumer-groups.sh에서 오프셋 리셋시 선택 가능한 옵션이 아닌 것은? 가장 처음(작은) 오프셋으로 리셋 가장 마지막(큰) 오프셋으로 리셋 특정 일시로 오프셋 리셋 마지막 메시지키 기준으로 리셋 특정 오프셋 기준으로 리셋 => 4 Q. kafka-consumer-groups.sh 에서 describe옵션으로 확인 할 수없는것은? 컨슈머 랙 토픽의 팔로우 파티션정보 마지막으로 읽은 레코드 오프셋 현재 레코드 오프셋 파티션 번호 컨슈머 정보(ID등) => 2 Q. Kafka consumer offset reset 할때 to-current 와 to-latest 의 차이점은? => to-latest : 마지막 commit offset 으로 To-current : 현재 처리하기 전 commit offset 으로