1. 설치

카프카를 설치하면 주키퍼를 실행하기 위한 설정 파일과 카프카를 설정하기 위한 설정 파일이 모두 존재하고 있어서

하나의 서버에서 카프카와 주키퍼를 모두 띄워서 사용할 수 있다.

하지만 주키퍼 전용 서버를 따로, 카프카 전용 서버를 따로 구성한다.(모든 서버는 centos 76)

 

아래와 같이 zookeeper 서버 3대, kafka 서버 3대로 구성한다.

zookeeper는 클러스터로 구성할 경우 과반수 이상의 서버가 정상일 때만 지속적인 서비스가 가능하기 때문에 홀수로 구성해야 한다.

따라서 3대를 설치한다.(from. twofootdog.tistory.com/89, judo0179.tistory.com/112)
카프카는 굳이 홀수로 구성할 필요는 없다.(from. gunju-ko.github.io/kafka/2018/12/21/%EC%B9%B4%ED%94%84%EC%B9%B4%EC%84%A4%EC%B9%98.html)

하지만 다수의 브로커로 클러스터를 구성하면

1. 처리량을 분산시켜 확장할 수 있는 장점이 있고,

2. 서버 장애에 따른 데이터 유실을 막기 위해 replication을 할 수 있다는 장점이 있다.

3. 또한 replication을 사용하면 현재 사용 중인 kafka 시스템을 중단시키지 않고 유지보수 작업을 수행할 수도 있다.

 

 

 

 

kafka를 설치하기 위해서는 zookeeper를 먼저 설치해야 한다.

kafka 아키텍쳐 내에서 kafka broker들을 하나의 클러스터로 코디네이팅하기 위해서는 별도의 분산 코디네이팅 시스템이 필요.

zookeeper는 kafka 클러스터의 설정 정보를 최신으로 유지하며 클러스터 내에서 broker 서버가 추가되거나 삭제될 때 그 정보를 클러스터에 속한 서버들끼리 공유할 수 있도록 한다.

클러스터 내의 broker 서버는 zookeeper에게 토픽 metadata를 공유하고, zookeeper는 kafka 클러스터의 리더를 발탁한다.

 

zookeeper 설치

1. jdk 설치

jdk 다운로드

# jdk를 설치할 서버로 가져온다.
$ curl -o OpenJDK11-jdk_x64_linux_hotspot_11_28.tar.gz https://download.java.net/java/ga/jdk11/openjdk-11_linux-x64_bin.tar.gz

# unzip
$ tar -xvzf OpenJDK11-jdk_x64_linux_hotspot_11_28.tar.gz

# 디렉토리 이름 변경
$ mv jdk-11+28 jdk


# 환경 변수 설정
# /etc/profile은 시스템 사용자 전체에 적용할 쉘 환경 변수를 담는다.
$ sudo vi /etc/profile // 해당 파일 열어서 아래 두 줄을 추가
export JAVA_HOME='/home/deploy/jdk' // jdk가 설치된 디렉토리
export PATH=$PATH:$JAVA_HOME/bin

자바 설치 확인

 

2. zookeeper 설치

# zookeeper를 설치할 서버로 가져온다.
$ curl -o apache-zookeeper-3.5.5-bin.tar.gz http://archive.apache.org/dist/zookeeper/zookeeper-3.5.5/apache-zookeeper-3.5.5-bin.tar.gz

# unzip
$ tar -xvzf apache-zookeeper-3.5.5-bin.tar.gz

# 이름 변경
$ mv apache-zookeeper-3.5.5-bin

설치하고 나면 아래와 같은 디렉토리가 생성

zookeeper 설치 후 생성된 디렉토리

주키퍼를 실행시키기 위해서는 아래와 같은 명령어를 실행한다.

# bin 디렉토리로 이동
$ ./zkServer.sh start

zkServer.sh를 들여다 보면

zkServer.sh

zkEnv.sh에서 어떤 값들을 실행시킨다.

zkEnv.sh를 들여다 보면

zkEnv.sh

zoo.cfg에 설정들을 가져오는 것을 알 수 있다.

따라서 conf 디렉토리 아래에 zoo.cfg 파일을 생성하여 설정해줘야 한다.(각 필드 설명은 여기서. engkimbs.tistory.com/560)

호스트 네임은 가림

data 디렉토리, log 디렉토리는 각자 생성해준다.

server.{myid}에서 myid는 클러스터 내에서 유니크해야 한다.

dataDir에 설정한 경로 myid 파일을 만들어서 idr값을 넣어준다.

myid

3. 실행

# bin 디렉토리로 이동
$ ./zkServer.sh start

# 프로세스가 잘 떳는지 확인
$ ps -ef | grep zookeeper

실행

4. 서비스로 등록

커맨드로 실행시키는 것이 아니라 서비스로 등록하여 데몬으로 떠잇도록 한다.

등록한 서비스를 통해 zookeeper를 실행

# kafka 실행
$ sudo systemctl start zookeeper.service

# 실행되었는지 확인
$ sudo systemctl status zookeeper.service

# 프로세스가 떳는지 확인
$ ps -ef | grep zookeeper

 

Service란?

Service

* 평상시에도 가동되는 서버의 프로세스

* DB서버, Web서버, FTP서버, WAS등..

* systemd(PID: 1)라는 서비스 매니저 프로그램으로 작동시키고 관리

* 명령어

  * systemctl start {service name} => 서비스 시작

  * systemctl stop {service name} => 서비스 중지

  * systemctl restart {service name} => 서비스 재시작

  * systemctl status {service name} => 서비스 상태 확인

  * systemctl enable {service name} => 부팅 시 서비스 자동 시작

  * systemctl disable {service name} => 부팅 시 서비스 자동 시작 해제

* centos의 경우 /lib/systemd/system/ 아래에 {service_name}.service의 형태로 서비스 script를 위치시킴.

* 스크립트는 아래와 같은 형태

* 이와 관련해서는 victorydntmd.tistory.com/215 여기에 설명이 잘 되어 있음. 

 

service로 등록하는 이유



Daemon

메모리에 상주하면서 특정 요청이 오면 즉시 대응할 수 있도록 대기중인 프로세스(systemd도 데몬)

* ssh 서버 데몬인 sshd처럼 데몬을 뜻하는 'd'를 이름 끝에 달고 있음.

* PPID는 1(systemd)

Service

* 평상시에도 가동되는 서버의 프로세스

* DB서버, Web서버, FTP서버, WAS등..

* systemd(PID: 1)라는 서비스 매니저 프로그램으로 작동시키고 관리

* 명령어

  * systemctl start {service name} => 서비스 시작

  * systemctl stop {service name} => 서비스 중지

  * systemctl restart {service name} => 서비스 재시작

  * systemctl status {service name} => 서비스 상태 확인

  * systemctl enable {service name} => 부팅 시 서비스 자동 시작

  * systemctl disable {service name} => 부팅 시 서비스 자동 시작 해제

* centos의 경우 /lib/systemd/system/ 아래에 {service_name}.service의 형태로 서비스 script를 위치시킴.

* 스크립트는 아래와 같은 형태

* 이와 관련해서는 victorydntmd.tistory.com/215 여기에 설명이 잘 되어 있음. 

 

service로 등록하는 이유

먼저 리눅스 서버나 클라이언트의 머신에서 부팅시 자동으로 실행되게 하고싶은 application이 있거나,
server application의 stop, restart등을 쉽게 관리하고싶을때 필요하다.

출처: https://khann.tistory.com/5 [khann's IT와 경제 블로그]



Daemon

메모리에 상주하면서 특정 요청이 오면 즉시 대응할 수 있도록 대기중인 프로세스(systemd도 데몬)

* ssh 서버 데몬인 sshd처럼 데몬을 뜻하는 'd'를 이름 끝에 달고 있음.

* PPID는 1(systemd)

 

 

kafka 설치

1. 카프카 설치전 JDK가 설치되어 있어야 함.

 

2. 카프카 설치

# kafka를 설치할 서버로 가져온다.
$ curl -o kafka_2.12-2.3.1.tgz https://archive.apache.org/dist/kafka/2.3.1/kafka_2.12-2.3.1.tgz

# unzip
$ tar -xvzf kafka_2.12-2.3.1.tgz

# 이름변경
$ mv kafka_2.12-2.3.1 kafka

unzip 후 만들어진 파일들

kafka는 실행하기 위해 zookeeper와 연결되어 있어야 한다.

따라서 config 디렉토리에 있는 server.properties 파일을 수정해서 먼저 설치한 주키퍼 클러스터와 연결해준다.

그리고 kafka 클러스터 구성을 위해 broker.id를 유니크하게 설정해준다.(zookeeper의 myid와 같이 1,2,3 ..)

# server.properties
...(생략)
broker.id=1
zookeeper.connect={ zookeeper_server_hostname:port }
...(생략)


ex)
kafkaserver01.co.kr:2181,kafkaserver02.co.kr:2181

설정 값 역할

broker.id

모든 kafka broker가 가지는 cluster내 고유한 값

 

port

kafka의 실행에 사용되는 TCP 포트

 

zookeeper.connect

broker의 metadata를 저장하기 위해 사용되는 zookeeper의 위치

 

log.dirs

kafka는 모든 message를 log segment 파일에 모아서 디스크에 저장하는데, 이 때의 위치.

여러 경로를 쉼표로 구분하여 지정할 수 있고, 두 개 이상의 경로가 지정되면 해당 broker가 모든 경로에 partition을 저장한다.

 

num.partitions

새로운 topic이 몇 개의 partition으로 생성되는지를 나타냄.

topic의 partition 개수는 증가만 가능하고 감소는 될 수 없다.

 

log.retention.ms

kafka가 얼마 동안 message을 보존할지를 나타냄.

 

log.retention.bytes

저장된 message들의 전체 byte크기를 기준으로 message를 얼마나 보존할지를 나타냄.

모든 partition에 적용된다.

예를 들어, 하나의 topic이 8개의 partition으로 되어 있고, log.retention.bytes의 값이 1GB로 설정되면 해당 topic에 보존되는 message들의 전체 크기는 최대 8GB가 된다.

 

log.segment.bytes

message 보존 설정(log.retention.ms, log.retention.bytes)은 개별 message가 아닌 log segment 파일을 대상으로 처리된다.

message가 kafka broker에 생성될 때는 해당 partition의 log segment 파일 끝에 추가된다.

log.segment.bytes에 설정된 크기가 되면 해당 log segment 파일이 닫히고 새로운 것이 생성되어 열린다.

예를 들어, 하루에 100MB의 message만 topic에 수록되고 log.segment.bytes의 값이 1GB로 설정되어 있다면, 하나의 log segment를 채우는데 10일이 소요될 것.

그러나 log segment 파일은 닫혀야만 만기가 될 수 있으므로 만약 log.retention.ms가 1주일로 설정되어 있다면
log segment에 저장된 messageemfdms 17일 동안 보존될 것.

왜냐하면 마지막 message를 쓰면서 log segment 파일이 닫히는데 10일이 소요되고, 다시 보존기간인 7일이 지나야 만기되기 때문.

 

log.segment.ms 

시간을 통해 log segment가 닫히는 시간을 조정

 

message.max.bytes

message의 최대 크기를 제한.

해당 크기보다 큰 message를 전송하려는 producerdprp는 broker가 에러를 보내고 message를 받지 않는다.

 

 

해당 어플리케이션(카프카)를 service에 등록한다

등록한 service를 통해 kafka를 실행시켜 본다.

# kafka 실행
$ sudo systemctl start kafka.service

# 실행되었는지 확인
$ sudo systemctl status kafka.service

정상적으로 실행되지 않았다면 아래와 같이 나옴.

왜 실행이 되지 않았는지 아래 명령어로 확인 할 수 있음.

$ sudo journalctl -xe

혹은 kafka 디렉토리 아래의 logs디렉토리에 있는 server.log 파일을 보면 어떤 에러가 발생해서 kafka가 실행되지 않았는지 알 수 있음.

 

정상적으로 실행되었다면 아래와 같이 active 상태가 됨.

 

 

kafka manager

zookeeper와 연동하여 kafka 클러스터 정보를 웹에서 편하게 볼 수있고, 설정을 변경할 수 있다.

 

1. 설치

kafka, zookeeper와 같이 파일을 찾아서 다운로드 후 압축을 풀어준다.

이후 config 디렉토리 아래에 있는 application.conf 파일 중 아래 부분을 수정해준다.

# 주키퍼 서버 목록 입력
kafka-manager.zkhosts="zookeeper_server1,zooker_server2,..."

 

2. 서비스 등록

[Unit]
Description=Kafka Manager
After=network.target
Requires=network.target

[Service]
Type=simple
User=deploy
Group=deploy
SyslogIdentifier=kafka
WorkingDirectory=/home/deploy/kafka-manager
Environment=JAVA_HOME=/home/deploy/jdk
Restart=no
TimeoutStartSec=10min
ExecStart=/home/deploy/kafka-manager/bin/kafka-manager -Dhttp.port=9000
SuccessExitStatus=143

[Install]
Alias=kafka-manager
WantedBy=multi-user.target

이후 웹 브라우저에서 {서버이름:9000}으로 접속하면 카프카 매니저에 접속가능하다.