데이터 파이프라인 설치 (1) - Kafka

카프카는 메세지 브로커 기능을 담당한다. 카프카가 무엇인지, 어떻게 기능하는지는 추후에 다른 글에서 다시 정리하도록 하고 여기서는 설치와 기본 설정에 대해서만 다루겠다.

설치 환경
- CPU : 4 core
- RAM : 8 GB
- DISK : 40 GB
- OS : ubuntu 20.04
- kafka version : 2.12-2.6.0

1. 다운로드

kafka 홈페이지에서 소스 파일을 다운로드 받으면 된다. 다만 버전 통일성을 위해 나는 동료에게 받은 소스파일을 사용해서 2.12 버전을 사용했다.

2. zookeeper 설치

카프카는 기본적으로 분산 시스템을 이루고있기 때문에 분산 코디네이션 시스템을 위해 주키퍼를 함께 사용한다.

다운받은 소스파일 압축을 풀면 zookeeper-server-*.sh 파일이 존재한다. 이 쉘파일을 실행시켜 주키퍼를 가동시키면 된다.

# zookeeper 실행
bin/zookeeper-server-start.sh -daemon config/zookeeper.properties

데몬 형태로 실행시키기 위해 -daemon 설정을 추가하든지, 서비스 파일을 구성해서 실행시킬 수도 있다.

sudo vi /etc/systemd/system/zookeeper.service

# zookeeper.service
[Unit]
Description=zookeeper
After=kafka.service

[Service]
Type=simple
User=ubuntu
Group=ubuntu
SyslogIdentifier=zookeeper
WorkingDirectory=/home/ubuntu/kafka
Restart=on-abnormal
RestartSec=0s
ExecStart=/home/ubuntu/kafka_2.12-2.6.0/bin/zookeeper-server-start.sh /home/ubuntu/kafka_2.12-2.6.0/config/zookeeper.properties
ExecStop=/home/ubuntu/kafka_2.12-2.6.0/bin/zookeeper-server-stop.sh

[Install]
WantedBy=multi-user.target

그리고 이번 데이터 파이프라인에서는 단일 노드로 구성하지만 클러스터로 분산 시스템을 설정할 때는 zookeeper broker 번호를 지정해주어야 한다.

zookeeper.properties를 보면 /tmp/zookeeper 경로가 데이터 경로이기 때문에 해당 경로에 myid라는 파일을 생성하고 번호를 넣어주면 된다. 단일 구성이지만 번호를 1로 지정해주었다.

echo 1 | tee -a /tmp/zookeeper/myid

그리고 zookeeper.properties에 클러스터 정보를 추가해준다.

vi config/zookeeper.properties

server.1=192.168.65.193:2888:3888

initLimit=5
initLimit=2

단일 노드이기 때문에 한 개만 적었다.

3. kafka 설치

주키퍼와 마찬가지로 소스파일에서 쉘 파일을 실행시키면 된다. 같은 방식이기 때문에 -daemon 설정이나 서비스 파일을 통해 실행시킨다.

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

sudo vi /etc/systemd/system/kafka.service

[Unit]
Requires=zookeeper.service

[Service]
Type=simple
User=ubuntu
Group=ubuntu
SyslogIdentifier=kafka
WorkingDirectory=/home/ubuntu/kafka
Restart=on-abnormal
RestartSec=0s
ExecStart=/home/ubuntu/kafka_2.12-2.6.0/bin/kafka-server-start.sh /home/ubuntu/kafka_2.12-2.6.0/config/server.properties
ExecStop=/home/ubuntu/kafka_2.12-2.6.0/bin/kafka-server-stop.sh
Restart=on-abnormal

[Install]
WantedBy=multi-user.target

카프카의 설정도 몇 가지 추가해주어야 한다.
config/server.properties에서 브로커 id와 listener 정보를 추가하면 된다.

vi config/server.properties

broker.id=1 # 위에서 1로 설정
listeners=PLAINTEXT://:9092
advertised.listeners=PLAINTEXT://{kafka1 IP}:9092

4. kafka 실행 및 테스트

# zookeeper 및 kafka 실행
systemctl enable zookeeper.service
systemctl start zookeeper.service

systemctl enable kafka.service
systemctl start kafka.service

# create topic
bin/kafka-topics.sh --create --bootstrap-server 192.168.65.193:9092 --topic test

test라는 토픽을 생성했다. --bootstrap-server는 브로커의 호스트이므로 현재는 주키퍼와 카프카가 동일 노드에 설치되었으므로 localhost:9092를 사용해도 무관하다.

프로듀서와 컨슈머가 동작하는지 확인

bin/kafka-console-producer.sh --bootstrap-server 192.168.65.193:9092 --topic test

프로듀서를 통해 메세지를 보낸다.

bin/kafka-console-consumer.sh --bootstrap-server 192.168.65.193:9092 --topic test

컨슈머에서 메세지를 잘 받아오는 것을 확인할 수 있다.