카프카 유용한 스크립트 목록

kafka-topics.sh

토픽 생성 명령어

bin/kafka-topics.sh --create --bootstrap-server my-kafka:9092 --topic hello.kafka

bin/kafka-topics.sh --create --bootstrap-server my-kafka:9092 --partitions 10 --replication-factor 1 --topic hello.kafka2 --config retention.ms=172800000

 

 

토픽 확인 명령어

bin/kafka-topics.sh --bootstrap-server my-kafka:9092 --topic hello.kafka --describe

Topic: hello.kafka    PartitionCount: 1    ReplicationFactor: 1    Configs: segment.bytes=1073741824
    Topic: hello.kafka    Partition: 0    Leader: 0    Replicas: 0    Isr: 0

 

 

토픽에 대한 파티션 추가 명령어

bin/kafka-topics.sh --bootstrap-server my-kafka:9092 --topic test --alter --partitions 10

카프카는 토픽 추가만 가능하다! (삭제 시 파티션 재분배의 복잡성 때문에 구현되지 않았다.)

kafka-config.sh

카프카 토픽 옵션 설정

 

 

--alter, --add-config 옵션을 사용해서 min.insync.replicas 옵션을 토픽별로 설정할 수 있다.

bin/kafka-configs.sh --bootstrap-server my-kafka:9092 --alter -add-config min.insync.replicas=2 --topic test

 

 

카프카 브로커 기본값 조회

 

--broker, --all, --describe 옵션을 사용해서 브로커에 설정된 각종 기본값을 조회할 수 있다.

bin/kafka-configs.sh --bootstrap-server my-kafka:9092 --broker 0 --all --describe

kafka-console-producer.sh

key가 null인 레코드를 추가하기

bin/kafka-console-producer.sh --bootstrap-server my-kafka:9092 --topic hello.kafka
> hello
> message
> 0
> 1
> 2
> (Ctrl + D)

Ctrl + D 로 EOF를 나타내 종료 가능하다.

 

key를 지정하여 레코드 추가하기

bin/kafka-console-producer.sh --bootstrap-server my-kafka:9092 --topic hello.kafka --property "parse.key=true" --property "key.separator=:"
> k1:message
> k2:hello
> k3:new
> k4:hi
> ...

kafka-console-consumer.sh

레코드를 확인하기 위해서 가장 많이 쓰는 명령어(쉘 스크립트)

 

레코드 확인하기

레코드의 메시지 키와 메시지 값을 확인하고 싶다면 --property 옵션을 사용하면 된다.
--max-messages 옵션을 사용하면 최대 컨슘 메세지 개수를 설정할 수 있다.
--partition 옵션을 사용하면 특정 파티션만 컨슘할 수 있다.

토픽의 모든 레코드 확인하기

bin/kafka-console-consumer.sh --bootstrap-server my-kafka:9092 --topic hello.kafka --from-beginning

 

 

토픽의 모든 레코드를 키와 함께 확인하기

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

 

 

토픽의 레코드 1개만 확인하기

bin/kafka-console-consumer.sh --bootstrap-server my-kafka:9092 --topic hello.kafka --from-beginning --max-messages 1

 

 

특정 파티션의 레코드 조회하기

bin/kafka-console-consumer.sh --bootstrap-server my-kafka:9092 --topic hello.kafka --from-beginning --partition 04

 

 

특정 컨슈머 그룹을 기준으로 레코드 조회하기

bin/kafka-console-consumer.sh --bootstrap-server my-kafka:9092 --topic hello.kafka --group hello-group --from-beginning

kafka-consumer-groups.sh

컨슈머 그룹은 따로 생성하는 명령을 날리지 않고 컨슈머를 동작할 때 컨슈머 그룹이름을 지정하면 새로 생성된다. 생성된 컨슈머 그룹의 리스트는 kafka-consumer-groups.sh 명령어로 확인할 수 있다.

--describe 옵션을 사용하면 해당 컨슈머 그룹이 어떤 토픽을 대상으로 레코드를 가져갔는지 상태를 확인할 수 있다. 파티션 번호, 현재까지 가져간 레코드의 오프셋, 파티션 마지막 레코드의 오프셋, 컨슈머 랙(LAG), 컨슈머 ID, 호스트를 알 수 있기 때문에 컨슈머의 상태를 조회할 때 유용하다.

컨슈머 랙이란? 프로듀서와 컨슈머간의 처리 속도 차이(지연)를 확인하는 지표

# 컨슈머 그룹의 리스트를 조회
❯ bin/kafka-consumer-groups.sh --bootstrap-server my-kafka:9092 --list
# 특정 컨슈머 그룹이 레코드를 조회해간 상태 확인
❯ 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-reassign-partitions.sh

리더 파티션과 팔로워 파티션이 위치를 변경할 수 있다. 카프카 브로커에는 auto.leader.rebalance.enable 옵션이 있는데 이 옵션의 기본값은 true로써 클러스터 단위에서 리더 파티션을 자동 리밸런싱하도록 도와준다.
브로커의 백그라운드 스레드가 일정한 간격으로 리더의 위치를 파악하고 필요시 리더 리밸런싱을 통해 리더의 위치가 알맞게 배분된다.

kafka-delete-record.sh

주어진 OFFSET과 이전의 모든 레코드를 지우는 명령어

kafka-dump-log.sh

카프카를 직접 운영할 때 사용할 수 있는 명령어

 

 

출처: 인프런 - [아파치 카프카 애플리케이션 프로그래밍] 개념부터 컨슈머, 프로듀서, 커넥트, 스트림즈까지! , 데브원영