Kafka 명령어

이번 포스트에서는 명령어를 통해 토픽을 생성, 수정하여 데이터를 전송(Producer)하고 받을 수(Consumer) 있는 명령어들을 살펴보겠습니다.

카프카의 명령어들은 관련 애플리케이션을 개발하고 클러스터를 운영할 때 자주 쓰이기 때문에 명령어를 손에 익혀두면 좋을것입니다.

 

1. Kafka-topics.sh

Topic과 관련된 명령어를 실행합니다.

Tip! 토픽을 생성하는 2가지 방법 1) Kafka Consumer 또는 Kafka Producer가 Kafka Broker에 생성되지 않은 토픽에 대해 데이터를 요청할 경우 2) 커맨드 라인 툴로 명시적으로 토픽을 생성하는 경우 토픽을 효과적으로 유지보수 하기 위해서는 2번 방법을 추천 -> 토픽마다 처리되어야 하는 데이터의 특성이 다르기 때문에

1) 토픽 생성 

--create 옵션을 사용하여 토픽을 생성할 수 있다.

# kafka-topics.sh --create --bootstrap-server localhost:9092 --topic {TOPIC-NAME}

① --create : 해당 옵션으로 토픽을 생성하는 명령어

② --bootstrap-server : 토픽을 생성할 카프카 클러스터를 구성하는 Broker들의 IP와 port를 작성

③ --topic : 토픽 이름을 작성. 토픽 이름은 내부 데이터가 무엇인지 유추가 가능할 정도로 자세히 작성

위와 같이 생성하면 Broker에 설정된 기본값으로 생성된다. (config/server.properties 파일에서 확인 가능)

# kafka-topics.sh --create --bootstrap-server localhost:9092 --partitions 3 --replication-factor 1 --config retention.ms=172800000 --topic {TOPIC-NAME}

① --partitions : 파티션 개수를 지정. 파티션 최소 개수는 1개

② --replication-factor : 토픽의 파티션을 복제할 복제 개수를 작성

• 1은 복제를 하지 않고 사용한다는 의미
• 2는 1개의 복제본을 사용하겠다는 의미
• 복제 개수의 최소 설정은 1이고 최대 설정은 통신하는 Kafka Cluster의 Broker 개수이다.
• 운영환경의 경우 3개 이상의 Kafka Broker를 운영하는 것이 일반적이라 2 또는 3으로 복제개수를 설정

③ --config : kafka-topics.sh 명령어에 포함되지 않는 추가적인 설정이 가능. retention.ms는 토픽의 데이터 유지 기간을 나타냄

 

2) 토픽 리스트 조회

--list 옵션을 사용하여 Kafka Cluster에 생성된 토픽들의 이름을 확인할 수 있다.

# kafka-topics.sh --bootstrap-server localhost:9092 --list

 

3) 토픽 상세 조회

--describe 옵션을 사용하여 생성된 토픽의 상태를 확인할 수 있다.

# kafka-topics.sh --bootstrap-server localhost:9092 --describe [--topic {TOPIC-NAME}] 

파티션 개수, 복제된 파티션이 위치한 Broker번호, 토픽이 현재 어느 Broker에 존재하는지 등 확인 가능하다.

 

2. Kafka-console-producer.sh

생성된 토픽에 데이터를 넣을 수 있는 명령어입니다.

토픽에 넣는 데이터는 '레코드(record)'라고 부르며 메시지 키(key)와 메세지 값(value)으로 이루어져 있습니다.

 

1) 메시지 키(key)없이 메세지 값(value)만 보내기

메세지 키(key)는 자바의 null로 기본 설정되어 Broker로 전송된다.

# kafka-console-producer.sh --bootstrap-server localhost:9092 --topic {TOPIC-NAME}

kafka-console-producer.sh로 전송되는 record 값은 UTF-8을 기반으로 Byte로 변환되고 ByteArraySerializer로만 직렬화된다.

즉, String이 아닌 타입으로는 직렬화 하여 전송할 수 없기 때문에 다른 타입으로 전송하고 싶다면 직접 카프카 프로듀서 애플리케이션을 개발해야 한다.

 

2) 메시지 키(key)를 가지는 record 전송하기

# kafka-console-producer.sh --bootstrap-server localhost:9092 --topic {TOPIC-NAME} --property "parse.key=true" --property "key.separator=:"

① parse.key : true로 두면 record를 전송할 때 메세지 키(key)를 추가할 수 있다.

② key.separator : 메세지 키(key)와 메세지 값(value)을 구분하는 구분자를 선언할 수 있다. key.separator를 선언하지 않으면 기본 설정은 Tab delimiter(\t)이다.

 

3. Kafka-console-consumer.sh

Producer를 이용해 토픽으로 전송한 데이터는 해당 명령어로 확인할 수 있다.

이때 필수 옵션으로 --bootstrap-server에 Kafka Cluster 정보, --topic에 토픽 이름이 필요하다.

# kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic {TOPIC-NAME} --from-beginning

① --from-beginning : 해당 토픽에 저장된 가장 처음 데이터부터 출력된다. (추가 옵션)

# kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic {TOPIC-NAME} --property print.key=true --property key.separator="-" --from-beginning

① print.key : 메시지 키를 확인할 수 있다.

② key.separator : 메세지 키 값을 구분하기 위해 사용할 수 있다. key.separator를 설정하지 않으면 tab delimieter(\t)가 기본값으로 설정되어 출력된다.

 

이상으로 Kafka에서 사용되는 기본적인 명령어를 보았습니다.

사용하다가 또 필요하다 싶은 명령어가 있다면 추가해서 작성해보겠습니다. 😌

 

😶 출처

아파치 카프카 프로그래밍 with 자바 (최원영 지음)