Kafka 설치 / Kafka와 Logstash 연동하기

- Kafka 구성도

- kafka : 분산 메시지 스트리밍 플랫폼

- 서버(=node,broker)를 여러대 둘 수 있고, 주키퍼 클러스터에 의해 관리됨.

- topic이라는 개념이 있어, 카프카에 메세지를 전달하거나 카프카의 메세지를 읽는 주체들은 이 topic을 key로써 카프카에 원하는 메세지를 구독할 수 있다.

- 카프카는 여러 토픽을 생성하여 토픽 내에 여러 메세지를 큐(Queue) 형태로 쌓을 수 있다.

 

- Kafka 설치

 1. Open JDK 설치

sudo apt install openjdk-8-jdk

 

2. Kafka 설치

wget https://archive.apache.org/dist/kafka/3.2.3/kafka_2.13-3.2.3.tgz
tar -xzf kafka_2.13-3.2.3.tgz 

cd kafka_2.13-3.2.3	// kafka 디렉토리로 이동

 

3. Kafka Config 세팅

# 데이터 디렉토리 생성
mkdir -p tmp/zookeeper		// 위치는 자유롭지만 난 보통 kafka 디렉토리 내에 생성함!

cd tmp/zookeeper
pwd 		// -----> 생성된 디렉토리로 이동하여 pull path 복사!
cd..
cd..  		// kafka 디렉토리 메인으로 이동

# 위 디렉토리 내에 노드 번호를 node_id파일에 기록한다.
echo 1 > tmp/zookeeper/node_id

# zookeeper 클러스터 설정
vi config/zookeeper.properties 

##################### 다음 내용 추가
dataDir=/home/ubuntu/kafka_2.13-3.2.3/tmp/zookeeper  # 위에서 생성한 데이터 디렉토리 full path
# 설정하는 노드의 IP에 해당하는 서버는 0.0.0.0 으로 설정한다.
### 노드가 하나인 경우는 server.1=0.0.0.0:2888:3888 만!!
server.1=0.0.0.0:2888:3888
#server.2={노드2번_IP}:2888:3888   # (서버)노드 추가 시
#server.3={노드3번3_IP}:2888:3888   # (서버)노드 추가 시

# the port at which the clients will connect
clientPort=2181
tickTime=2000
initLimit=10
syncLimit=5
####################

# kafka 클러스터 설정
vi config/server.properties

#################### 다음 내용 추가
broker.id=1

# 카프카 브로커(=서버, =노드)가 내부적으로 바인딩하는 주소
listeners=PLAINTEXT://0.0.0.0:9092

# 카프카 producer, consumer에게 노출할 주소. 설정하지 않을 경우 디폴트로 listeners 설정을 따른다
advertised.listeners=PLAINTEXT://{호스트IP}:9092
log.dirs={kafka디렉토리}/tmp/log   	// log 쌓을 경로 설정

#### 로컬 노드 하나만 추가하는 경우, {노드1번 IP}에 host_IP 를 넣어주고 나머지는 지운다.
# zookeeper.connect=localhost:2181
zookeeper.connect={노드1번 IP}:2181,{노드2번 IP}:2181,{노드3번 IP}:2181
####################

 

4. Kafka 실행 / 중지

# 항상 zookeeper를 먼저 실행해야 한다.
sudo bin/zookeeper-server-start.sh config/zookeeper.properties	// 일반실행
sudo bin/zookeeper-server-start.sh -daemon config/zookeeper.properties	// 데몬실행

# new Terminal 오픈하여 Kafka 실행
sudo bin/kafka-server-start.sh config/server.properties	// 일반실행
sudo bin/kafka-server-start.sh -daemon config/server.properties	// 데몬실행


# Kafka, zookeeper 중지
bin/kafka-server-stop.sh
bin/zookeeper-server-stop.sh

 

5. Topic 관리

## 단일 노드
# topic 생성
bin/kafka-topics.sh --create --topic <topic명> --bootstrap-server localhost:9092
# topic 목록
bin/kafka-topics.sh --list --bootstrap-server localhost:9092
# topic 삭제
bin/kafka-topics.sh --delete --bootstrap-server localhost:9092 --topic <topic명>
# 전체 topic 상세 조회
bin/kafka-topics.sh --describe --bootstrap-server localhost:9092

## 복수 노드
# topic 생성  
bin/kafka-topics.sh --create --zookeeper {노드1번_ip}:2181,{노드2번_ip}:2181,{노드3번_ip}:2181 --replication-factor 3 --partitions 1 --topic <topic명>
# topic 삭제
bin/kafka-topics.sh --delete --zookeeper {노드1번_ip}:2181,{노드2번_ip}:2181,{노드3번_ip}:2181 --topic <topic명>
# topic 목록 확인
bin/kafka-topics.sh --list --zookeeper {노드1번_ip}:2181,{노드2번_ip}:2181,{노드3번_ip}:2181
# 전체 topic의 상세정보 확인
bin/kafka-topics.sh --describe --zookeeper {노드1번_ip}:2181,{노드2번_ip}:2181,{노드3번_ip}:2181

 

6. kafka publish(topic에 메시지 받기)/subscripe(topic 메시지를 읽을 수 있도록 열기) 테스트

  1) producer : 지정된 topic에 메세지를 ??

  2) consumer : 지정된 topic에 메세지를 보내면 kafka가 로컬에 출력

### 단일 노드
# publish
bin/kafka-console-producer.sh --bootstrap-server localhost:9092 --topic {topic명}

# subscripe (consumer)
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic {topic명}


### 복수 노드
# publish (producer)
bin/kafka-console-producer.sh --broker-list {노드1번ip}:9092,{노드1번ip}:9092,{노드1번ip}:9092 --topic <topic명>

# subscripe (cunsumer)
bin/kafka-console-consumer.sh --bootstrap-server {노드1번ip}:9092,{노드1번ip}:9092,{노드1번ip}:9092 --topic <topic명>

 

- Kafka 와 Logstash 연동하기

0. Logsatsh 설치 및 실행 방법

Linux(Ubuntu)에서 Logstash, Metricbeat 설치 / 실행

 

1. Kafka 실행

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

# new Terminal 하여 Kafka 실행
sudo bin/kafka-server-start.sh -daemon config/server.properties

 

2. Kafka 토픽 생성  ( test 토픽명 : kafka-test)

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

 

3. Logstash 설정 변경

# logstash 디렉토리 위치로 이동
vi config/{설정파일명}.conf    # 본인이 생성한 .conf 파일 편집

###############################

input {
  stdin{}
  beats {
    port => 5044
	}
}

output {
  stdout {}
  kafka {
    codec => json
    topic_id => "kafka-test"		# 메세지를 보낼 토픽명		
    bootstrap_servers => "{노드 호스트IP 192.168.x.x}:9092"		# kafka 브로커(노드) 주소. 복수개일 경우 ','로 구분하여 적어주기
  }
}

###############################

 

4. Logstash 실행 및 메세지 보내기

# Logstash 디렉토리 위치로 이동
bin/logstash -f /usr/share/logstash/{설정파일명}.conf

# Logstash가 정상 실행 되면, 같은 터미널 위치에 아무 문자 입력해보기
Hellow World!

Logstash 실행 후 'Hellow World' 입력.

Logstash에서 보낸 메세지를 받은 Kafka 모습