Apache Kafka [4] - 명령어

 

* 카프카 PATH로 cd하여 실행함을 가정한다.

실행중지

[ 카프카 실행 ]

./bin/kafka-server-start.sh ./config/server.properties --daemon

[ 카프카 중지 ]

./bin/kafka-server-stop.sh

토픽

[ 토픽생성 ]

./bin/kafka-topics.sh --create --zookeeper 172.123.123.111:2181, 172.123.123.222:2181, 172.123.123.333:2181 --replication-factor 3 --partitions 3 --topic kafka-topic-name

: 카프카 서버3대 (broker)에 복제를 3번하는 토픽을 만든다.

[ 토픽확인 ]

./bin/kafka-topics.sh --bootstrap-server kafkahost_IP:9092 --list

or

./bin/kafka-topics --zookeeper 172.123.123.111:2181 --topic kafka-topic-name --describe

or

./bin/kafka-topics.sh --list --zookeeper 172.123.123.111:2181

[ 지노드확인 ]

zookeeperPath/bin/zkCli.sh 

: 주키퍼접속

zk] ls /hadoop-kafka/brokers/ids 

: znode를 통해 확인

프로듀서/컨슈머

[ Producer messages 보내기 ]

./bin/kafka-console-producer.sh --broker-list 172.123.123.111:9092 --topic kafka-topic-name

[ Producer에 파일로 message 보내기 ]

./bin/kafka-console-producer.sh --broker-list 172.123.123.111:9092 --topic kafka-topic-name < /home/ubuntu/sample.json

[ Consume messages 받기 ]

./bin/kafka-console-consumer.sh --bootstrap-server 172.123.123.111:9092 --topic kafka-topic-name

컨슈머그룹

[ Consumer group 생성 ]

./bin/kafka-console-consumer.sh --bootstrap-server 172.123.123.111:9092 --topic kafka-topic-name --consumer-property group.id=kafka-group-id

[ Comsumer groupID 확인 ]

./bin/kafka-consumer-groups.sh --bootstrap-server 172.123.123.111:9092 --list

[ Comsumer group 삭제 ]

./bin/kafka-consumer-groups.sh --zookeeper 172.123.123.111:2181 --delete --group kafka-group-id

[ Consumer group상태 및 offset 확인 ]

./bin/kafka-consumer-groups.sh --bootstrap-server 172.123.123.111:9020 --group kafka-group-id --describe

설정

[ 로그 보관 주기설정 ]

./bin/kafka-topics.sh --zookeeper 172.123.123.111:2181 --alter --topic kafka-topic-name --config retention.ms=86400000 (초단위 10일)

: 카프카의 디스크 공간이 부족하다면 로그보관주기설정을 짧게하여 retention.ms값을 낮추는 것이다.

[ 로그 보관 주기변경확인 ]

./bin/kafka-topics.sh --zookeeper 172.123.123.111:2181 --describe --topic kafka-topic-name