kafka cluster 구성

2020/05/13 - [MessageQueue/Kafka] - kafka 특징

2020/05/13 - [MessageQueue/Kafka] - kafka 설치 및 clustering

2023.06.09 - [MessageQueue/Kafka] - kafka 설치 (Single mode)

 

1. kafka 최신 파일 다운로드

kafka는 단독으로 동작되지 않고 zookeeper에 의해 관리 되기 때문에 zookeeper를 함께 설치 하여하 한다. 기본적으로 kafka를 다운받으면 기본적으로 검증된 zookeeper 가 들어 있기때문에 다운 받지 않고 패키지 않에 들어 있는 zookeeper를 사용한다.

 

wget <http://apache.mirror.cdnetworks.com/kafka/2.2.1/kafka_2.12-2.2.1.tgz>

tar -xvzf kafka_2.12-2.2.1.tgz

cd kafka_2.12-2.2.1

 

2. Zookeeper 설정

각 인스턴스에 설치된 Kafka의 config/zookeeper.properties 파일은 하나의 Zookeeper를 실행하는데 쓰이는 설정 파일이다. 이 말은 zookeeper1.properties, zookeeper2.properties, ookeeper3.properties 이런식으로 여러개의 설정파일을 만들고 하나의 장비에서 다중으로 실행할 수 있다는 의미이다.

설정파일을 다음과 같이 3대의 서버에 동일하게 추가하자.

 

vi config/zookeeper.properties

# the directory where the snapshot is stored.

dataDir=**/tmp/zookeeper**

# the port at which the clients will connect

clientPort=2181

# disable the per-ip limit on the number of connections

# since this is a non-production config

maxClientCnxns=0

initLimit=5

syncLimmit=2

 

server.1=192.168.0.202:2888:3888

server.2=192.168.0.193:2888:3888

server.3=192.168.0.208:2888:3888

새로 추가한 설정값은 클러스터를 구성하는데 필요한 설정 값들안데 여기서 주의할 점은 모든 Zookeeper 서버들은 동일한 변수 값을 가지고 있어야 한다.

2.1 zookeeper properties 옵션 정보

주키퍼 기본 옵션 정보

2.2.1 initLimit

팔로워가 리더와 초기에 연결하는 시간에 대한 타임아웃 설정

2.2.2 syncLimit

• 팔로워가 리더와 동기화 하는데에 대한 타임아웃.
• 즉 이 틱 시간안에 팔로워가 리더와 동기화가 되지 않는다면 제거 된다.
• 이 두값은 dafault 기본값이 없기 때문에 반드시 설정해야 하는 값이다.그리고 server.1,2,3의 숫자는 인스턴스 ID이다.

2.2. myid 설정

• zookeeper 설치후 myid 파일을 생성하여 각 서버의 고유 ID를 설정한다.
• myid 명시 파일은 dataDir=/tmp/zookeeper 폴더에 생성한다.

#server-(1) (192.168.0.202)
mkdir /tmp/zookeeper
echo 1 > /tmp/zookeeper/myid


#server-(2) (192.168.0.193)
mkdir /tmp/zookeeper
echo 2 > /tmp/zookeeper/myid


#server-(3) (192.168.0.208)
mkdir /tmp/zookeeper
echo 3 > /tmp/zookeeper/myid

3. Kafka 설정

Kafka의 config/server.properties 파일은 하나의 Kafka를 실행하는데 쓰이는 설정 파일이다. Zookeeper와 마찬가지로 여러개의 설정파일을 만들고 다중 실행을 할 수 있다. 설정파일 config/server.properties에 3대 서버 각 환경에 맞는 정보를 입력해 준다. (클러스터링 구성시 기본 3대구성)

> vim config/server.properties

 

#192.168.0.202 개발서버 1번 IP
#192.168.0.193 개발서버 2번 IP
#192.168.0.208 개발서버 3번 IP

##########################
#서버 1 
##########################
broker.id=1
listeners=PLAINTEXT://:9092
advertised.listeners=PLAINTEXT://192.168.0.202:9092

#서버1호스트명:서버1포트,서버2호스트명:서버2포트,서버3호스트명:서버2포트/주기퍼노드명
zookeeper.connect=192.168.0.202:2181, 192.168.0.193:2181, 192.168.0.208:2181

##########################
#서버 2
##########################
broker.id=2
listeners=PLAINTEXT://:9092
advertised.listeners=PLAINTEXT://192.168.0.193:9092

#서버1호스트명:서버1포트,서버2호스트명:서버2포트,서버3호스트명:서버2포트/주기퍼노드명
zookeeper.connect=192.168.0.202:2181, 192.168.0.193:2181, 192.168.0.208:2181

##########################
#서버 3
##########################
broker.id=3
listeners=PLAINTEXT://:9092
advertised.listeners=PLAINTEXT://192.168.0.208:9092

#서버1호스트명:서버1포트,서버2호스트명:서버2포트,서버3호스트명:서버2포트/주기퍼노드명
zookeeper.connect=192.168.0.202:2181, 192.168.0.193:2181, 192.168.0.208:2181

 

zookeeper.connect 에서 마지막 /ykkim-kafka는 주기퍼노드명이다. 않으면 주기퍼 루트노드에 저장된다. 그렇게 되면 관리하기가 어려우므로 이렇게 별도로 노드명을 기재해준다.

3.1 zookepper 및 kafka 실행

# 주키퍼 실행
nohup bin/zookeeper-server-start.sh config/zookeeper.properties > /dev/null &

# 카프카 실행
nohup bin/kafka-server-start.sh config/server.properties > /dev/null &

#주키퍼 포트확인
netstat -ntlp | grep 2181

#카프카 포트확인
netstat -ntlp | grep 9092

#log 확인 (zookeeper, kafka)
tail -f /logs/server.log

 

4. topics 관리

4.1 topic 생성

 

#gamelog topoc 
bin/kafka-topics.sh --create --zookeeper 192.168.0.202:2181, 192.168.0.193:2181, 192.168.0.208:2181 --replication-factor 3 --partitions 1 --topic GameLog

#advertising topic 
bin/kafka-topics.sh --create --zookeeper 192.168.0.202:2181, 192.168.0.193:2181, 192.168.0.208:2181 --replication-factor 3 --partitions 3 --topic advertising

#example (standalone - test-topic)
bin/kafka-topics.sh --create --zookeeper  192.168.0.202:2181 --replication-factor 1 --partitions 1 --topic test-topic

 

4.2 Topic option

• –create : 토픽 생성
• –topic : 생성할 토픽명,
• –zookeeper : 주키퍼가 실행중인 호스트명
• –partitions : 생성할 토픽의 파티션 수
• –replication-factor : 생성항 토픽의 복사본 수 (cluster 가 떠있는 갯수 만큼 MAX)

4.2 Topic 리스트 확인

 

bin/kafka-topics.sh --list --zookeeper 192.168.0.202:2181, 192.168.0.193:2181, 192.168.0.208:2181

 

4.3 Topic 삭제

 

bin/kafka-topics.sh --delete --zookeeper 192.168.0.202:2181, 192.168.0.193:2181, 192.168.0.208:2181 --topic GameLog
bin/kafka-topics.sh --delete --zookeeper 192.168.0.202:2181, 192.168.0.193:2181, 192.168.0.208:2181 --topic advertising

# This will have no impact if delete.topic.enable is not set to true.
# 메세지가 다음과 같이 뜨는 경우 토픽 삭제에 대한 설정을 트루로 설정 해주어야 한다.
> cat config/server.properties | grep delete.topic
> vim config/server.properties
    ----------------------------------------------
    ...
    delete.topic.enable = true
    ...
    ----------------------------------------------

 

 

4.4 Topic 상세 정보 확인

 

# 토픽 상세 정보 확인 리플리케이션된 서버의 모든 토픽 정보
bin/kafka-topics.sh --describe --zookeeper 192.168.0.202:2181, 192.168.0.193:2181, 192.168.0.208:2181 --topic advertising

bin/kafka-topics.sh --describe --zookeeper 192.168.0.96:2181

5. 메세지 생성 및 소비

5.1 Producer 메세지 생성

5.1.1 example (cluster)

 

bin/kafka-console-producer.sh --broker-list 192.168.0.202:9092,192.168.0.193:9092,192.168.0.208:9092 --topic GameLog

 

5.1.2 example (standalone )

 

bin/kafka-console-producer.sh --broker-list 192.168.0.202:9092 --topic test-topic
bin/kafka-console-producer.sh --broker-list 192.168.0.96:9092 --topic test-topic

## option 설명 
# topic : 메세지를 가져올 토픽 
# broker-list: kafka가 실행중인 호스트를 적어준다.

 

5.2 Consumer 메시지 소비하기

5.2.1 example (cluster)

 

# 게임로그 토픽 소비
bin/kafka-console-consumer.sh --bootstrap-server 192.168.0.202:9092,192.168.0.193:9092,192.168.0.208:9092 --topic GameLog --from-beginning

# 광고 토픽 소비 
bin/kafka-console-consumer.sh --bootstrap-server 192.168.0.202:9092,192.168.0.193:9092,192.168.0.208:9092 --topic advertising --from-beginning

 

5.2.2 Standalone 예제

 

bin/kafka-console-consumer.sh --bootstrap-server 192.168.0.202:9092 --topic test-topic --from-beginning

 

5.3 테스트

• Producer Console 에서 메세지를 입력한다.
• Consumer console 에서 생산된 메세지들이 소비되는 것을 볼수 있다.

6. service 등록

6.1 zookeeper Service 등록

kafka embeded 를 이용하면 clusting 사용시 제대로 동작 되지 않는다. 실제 운영할때는 zookeeper 를 별도로 설치 하여 사용하길 권장 한다.

어떤이는 임베디드 되어 있는 zookeeper는 검증된 것이기때문에 이것을 사용하라고도 한다. system service 등록후 zookeeper를 사용하면 클러스트링이 잘안되는 문제가 있다,

nohup bin/zookeeper-server-start.sh config/zookeeper.properties & << 사용할것

6.2 주키퍼 서비스 등록 설정

vi /etc/systemd/system/zookeeper.service

[Unit]
Description=zookeeper
After=network.target

[Service]
Type=forking
User=mezzo
Group=mezzo
SyslogIdentifier=zookeeper
WorkingDirectory=/home/mezzo/message-queue/kafka_2.12-2.2.1/
Restart=always
RestartSec=0s
ExecStart=/home/mezzo/message-queue/kafka_2.12-2.2.1/bin/zookeeper-server-start.sh /home/mezzo/message-queue/kafka_2.12-2.2.1/config/zookeeper.properties
ExecStop=/home/mezzo/message-queue/kafka_2.12-2.2.1/bin/zookeeper-server-stop.sh

[Install]
WantedBy=multi-user.target

6.3 kafaka Service 등록 설정

nohup bin/kafka-server-start.sh config/server.properties &

vi /etc/systemd/system/kafka.service

 

[Unit]
Description=kafka
After=network.target

[Service]
Type=simple
User=mezzo
Group=mezzo
SyslogIdentifier=kafka
WorkingDirectory=/home/mezzo/message-queue/kafka_2.12-2.2.1/
Restart=always
RestartSec=0s
ExecStart=/home/mezzo/message-queue/kafka_2.12-2.2.1/bin/kafka-server-start.sh /home/mezzo/message-queue/kafka_2.12-2.2.1/config/server.properties
ExecStop=/home/mezzo/message-queue/kafka_2.12-2.2.1/kafka-server-stop.sh

[Install]
WantedBy=multi-user.target

7. 데몬 등록 및 시작

• 서비스 데몬 재시작
  • systemctl daemon-reload
• 주키퍼/카프카 실행
  • systemctl start zookeeper.service
  • systemctl start kafka.service
• 실행 상태 확인
  • systemctl status zookeeper.service
  • systemctl status kafka.service
• 시스템 부팅할때 자동실행 설정
  • systemctl enable zookeeper.service
  • systemctl enable kafka.service
• 서비스 로그 보기
  • journalctl -u ${serviceName}
  • journalctl -xe

reference

• https://kafka.apache.org/quickstart
• http://zookeeper.apache.org/doc/r3.3.3/zookeeperAdmin.html#sc_zkMulitServerSetup
• https://devtimes.com/bigdata/2019/01/25/kafka-cluster/
• https://www.joinc.co.kr/w/man/12/Kafka/docker
• https://yookeun.github.io/kafka/2018/07/01/kafka-install/

8. Kafka Topic 정리

8.1 Topic 과 partition

• 메세지는 토픽으로 분류되며 , 토픽은 여러개의 파티션으로 생성된다.
• 파태션 내의 한칸은 로그라 불리 운다.
• 데이터는 한칸의 로그에 순차적으로 추가되며 메세지의 상대적인 위치를 나타내는 것이 offset 이다.
• 파티션을 추가하게 되면 라운드 로빈 방식으로 Task가 추가 된다.
  • task 가 partition 0,1,2,3 에 순차적으로 쌓인다.

※ 주의 파티션은 늘리긴 쉬우나. 운영중에 축소할수는 없다.

8.2 Topic Gorup Id - Consumer 관계

• 컨슈머에서 같은 그룹 아이디의 경우 순차적으로 offset 위치를 사용하며, 소비한다.
• 그룹아이디가 다른 컨슈머는 offset 위치가 처음부터 다시 재조정 된다.
  • 여기서 눈치 챌수 있는데, 다른 컨슈머에의해 토픽의 Task 가 실행 되어도 실제 삭제 되지는 않는다.
  • 그룹 아이디가 다른경우 0번부터 데이터를 다시 소비 한다.
• Topic 의 task 가 유지되는 시간과 갯수는 환경설정에서 조정 한다.