아파치 카프카(Apache Kafka)는 아파치 소프트웨어 재단이 스칼라로 개발한 오픈 소스 메시지 브로커 프로젝트이다. 이 프로젝트는 실시간 데이터 피드를 관리하기 위해 통일된, 높은 처리량, 낮은 지연시간을 지닌 플랫폼을 제공하는 것이 목표이다. 요컨대 분산 트랜잭션 로그로 구성된, 상당히 확장 가능한 pub/sub 메시지 큐로 정의할 수 있으며, 스트리밍 데이터를 처리하기 위한 기업 인프라를 위한 고부가 가치 기능이다.

큐 (Queue)

먼저 큐에 대한 이해가 필요한데, 말 그대로 ‘기다리는 줄(열)’, ‘대기 열’을 생각하면 된다. 전공자라면 자료구조, 알고리즘 과목 등에서 접하게 되는데 Stack 과 함께 배운다.
FIFO¹ 구멍이 두 개인 파이프의 한쪽에 동전을 넣는다고 생각하면 된다. 계속 동전을 밀어 넣다 보면 가장 먼저 넣었던 동전부터 동전이 밀려나오는 구조가 된다.
이와 반되 되는 것이 Stack. 쌓는다는 의미의 stack으로 FILO² 컵에 동전을 쌓는 것을 상상하면 된다. 컵에 동전을 쌓아 넣다다가 동전을 꺼내면 나중에 넣었던 동전을 먼저 꺼내야 한다.

Message Queue

이렇게 먼저 넣은 값이 먼저 출력되는 구조를 이용해 메시지³를 전송하는 것이 Message Queue이다. Rabbit MQ, IBM MQ, Apache active MQ , Rocket MQ 등 다양한 종류의 Message Queue가 존재한다.
기본적인 개념은 모두 같지만 메시지를 주고 받는 방식에 따라 크게 두 종류로 구분할 수 있다.
– Push : 서버가 메시지를 클라이언트에 보내주는 방식
– Pull : 클라이언트가 메시지를 서버에서 가져오는 방식

별것 아닌 것 같지만 Pull 과 Push는 큰 차이를 가진다.

종류	Push	Pull
장점	· 메시지 발생 즉시 클라이언트에 전달	· 필요한 클라이언트만 메시지를 가져가므로 서버 부하 감소 · 네트워크 부하 감소
단점	· 서버가 모든 클라이언트에 연결해야 하므로 서버 부하 증가	· 메시지를 전달한 클라이언트 확인 불가
주 용도	카카오톡 등 메시지, 알림 전달	분석용 서버 로그 데이터 등

Push 방식과 Pull 방식 비교

Apache Kafka

Apache Kafka는 Pull 방식의 Message Queue이다.
중요한 개념으로 메시지를 발생 시키는 (또는 서버의 Queue에 메시지를 보내는) Producer(또는 Publisher)와 메시지를 가져가는 Consumer(또는 subscriber)가 있다.⁴

즉, 누군가 메시지를 보내는 곳, 누군가 메시지를 꺼내갈 곳이 Kafka 이다. ⁵

이 Kafka Broker가 하는 일과 핵심적인 개념은 다음과 같다.

Queue : 가장 핵심적인 기능이다. 수신된 메시지를 저장한다. kafka가 인지하는 형식으로 디스크에 저장된다.
Offset 관리 : 프로듀서가 보낸 메시지와 컨슈머가 가져간 메시지의 offset, 즉, 몇 개의 메시지가 수신되었고 어떤 클라이언트가 어디서부터 어디까지의 메시지를 가져갔는지 기억하고 관리한다.
Topic : Topic은 앞서 설명한 메시지를 담는 파이프 하나를 떠올리면 된다. 메시지를 넣을 파이프를 만들고 이름을 붙여서 관리한다. 이 Topic은 여러개의 Pratition으로 나뉠 수 있다.
Replica : 데이터의 복제, Broker나 디스크의 장애 등에 대비해 데이터를 복제하는 기능이다.

Public cloud에서

AWS : MKS(Managed Kafka Service)
GCP : Pub/Sub (Publisher와 Subscriber)
AZURE: HDInsight Kafka

2020-09-012022-05-06

Zookeeper #2. 설치와 설정

Zookeeper #1. 개요
Zookeeper #2. 설치와 설정
Zookeeper #3. 구동과 확인

Kafka #1. 개요
Kafka #2. 설치

*대충.. 이렇게 구성 해보자..*
*2888은 leader만 오픈 한다는 것을 기억하자.*

1. 다운로드 및 압축 해제
– Apache zookeeper 홈페이지에서 필요한 버전을 선택하거나 여기를 클릭하여¹다운로드 한다.
– 다운로드한 파일을 FTP 또는 SFTP등을 이용하여 서버에 업로드한다.
– 또는 wget 명령을 이용하여 서버에서 다운로드 한다.

[kafka@kafka ~]$ wget http://apache.tt.co.kr/zookeeper/zookeeper-3.5.8/apache-zookeeper-3.5.8-bin.tar.gz
--2020-09-01 08:27:41--  http://apache.tt.co.kr/zookeeper/zookeeper-3.5.8/apache-zookeeper-3.5.8-bin.tar.gz
Resolving apache.tt.co.kr (apache.tt.co.kr)... 211.47.69.77
Connecting to apache.tt.co.kr (apache.tt.co.kr)|211.47.69.77|:80... connected.
HTTP request sent, awaiting response... 200 OK
Length: 9394700 (9.0M) [application/x-gzip]
Saving to: ‘apache-zookeeper-3.5.8-bin.tar.gz’

100%[========================================================================>] 9,394,700   72.0KB/s   in 2m 6s

2020-09-01 08:29:47 (72.8 KB/s) - ‘apache-zookeeper-3.5.8-bin.tar.gz’ saved [9394700/9394700]

[kafka@kafka ~]$ ll
합계 91272
-rw-r--r--  1 kafka kafka 12436328  5월  1 04:53 apache-zookeeper-3.6.1-bin.tar.gz
drwxrwxr-x 76 kafka kafka     4096  9월  1 08:29 data
drwxr-xr-x  7 kafka kafka      245  8월 13 07:54 jdk1.8.0_251
lrwxrwxrwx  1 kafka kafka       16  8월 12 09:00 kafka -> kafka_2.13-2.6.0
drwxr-xr-x  9 kafka kafka      182  8월 13 09:48 kafka_2.13-2.6.0
-rw-r--r--  1 kafka kafka 65537909  8월  5 07:01 kafka_2.13-2.6.0.tgz
drwxrwxr-x  2 kafka kafka        6  8월 12 09:07 perl5

– 압축을 해제하고 심볼릭 링크를 생성한다.² ³

[kafka@kafka ~]$ tar -xvzf apache-zookeeper-3.6.1-bin.tar.gz
apache-zookeeper-3.6.1-bin/docs/
apache-zookeeper-3.6.1-bin/docs/skin/
    :
중략
   :
apache-zookeeper-3.6.1-bin/lib/metrics-core-3.2.5.jar
apache-zookeeper-3.6.1-bin/lib/snappy-java-1.1.7.jar
[kafka@kafka ~]$ ln -s apache-zookeeper-3.6.1-bin zookeeper
[kafka@kafka ~]$ ll
합계 12148
drwxrwxr-x 6 kafka kafka      133  9월  1 08:35 apache-zookeeper-3.6.1-bin
-rw-r--r-- 1 kafka kafka 12436328  9월  1 08:34 apache-zookeeper-3.6.1-bin.tar.gz
lrwxrwxrwx 1 kafka kafka       26  9월  1 09:12 zookeeper -> apache-zookeeper-3.6.1-bin

2. 환경변수 설정
※ /home/kafka 디렉토리에 압축을 해제했고 kafka 계정을 사용하는 것으로 가정한다.
– ~/.bash_profile 에 환경 변수를 등록한다. ⁴

[kafka@kafka zookeeper]$ vi ~/.bash_profile

export JAVJA_HOME=/home/kafka/jdk1.8.0_251
export PATH=$PATH:$JAVA_HOME/bin
 # JAVA_HOME 디렉토리 관련 설정. 

#여기부터 추가한다.
export ZOOKEEPER_HOME=/home/kafka/zookeeper
export ZOOBINDIR=$ZOOKEEPER_HOME/bin
 # zookeeper home 과 bin 설정
 # $ZOOKEEPER_HOME/bin/zkEnv.sh 에 $ZOOBINDIR이 명시되어있긴 하지만
 #편의를 위해 등록한다. zkEnv.sh 를 수정해도 된다.

export PATH=$PATH:$ZOOBINDIR
 #zookeeper bin 디렉토리를 PATH에 등록

 #변수를 별도록 export 하지 않으면 해당 변수는 '현재의 스크립트에서만 유효하다'

[kafka@zookeeper-kafka zookeeper]$ source ~/.bash_profile

3. zookeeper 설정
– 설정 파일 수정, node id 파일 생성, 클러스터간 설정 파일 동기화 세가지 작업이 필요하다.

가. $ZOOKEEPER_HOME/conf/zoo.cfg 파일을 생성하고 파일을 수정한다.

[kafka@kafka ~]$ vi $ZOOKEEPER_HOME/conf/zoo.cfg

tickTime=2000
 # 단위는 밀리초. 신호를 보내고 응답이 오기까지 설정된 시간만큼 대기한다.

initLimit=10
 # election 과정 이후 리더로 선출 된 zookeeper와 follower zookeeper들 간의 
 # 동기화 등에 사용되는   tick 시도 횟수. 즉 TickTime=2000에 initLimit=10일경우 
 # 총 20초동안 대기한다.
 # 관리하는 데이터의 양이 많을 경우 값을 늘려준다.

syncLimit=5
 # follower와 zookeeper 간의 동기화를 위한 틱 시도 횟수. 

dataDir=/home/kafka/zookeeper/datadir
 # zookeeper 데이터 디렉토리. tmp는 권장하지않는다.

clientPort=2181
 # zookeeper로 관리되는 클라이언트가 zookeeper에 연결 할 때 사용하는 포트

maxClientCnxns=7
# 최대 접속 가능한 클라이언트 갯수.
# 관리 대상 클라이언트가 많으면 값을 늘려준다.

autopurge.snapRetainCount=3
 # dataDir 디렉토리에 유지할 snapshot 갯수

autopurge.purgeInterval=1
 # 정리(삭제) 시도 주기, 간격. 
 # 단위는 시간.

server.1=192.168.100.111:2888:3888;2181
server.2=zookeeper2:2888:3888;2181
server.3=zookeeper3.fqdn.name:2888:3888;2181
 # 노드 정보. IP, host명(hosts 파일에 있는), FQDN 을 사용할 수 있다.
 # 2888은 leader노드가 follower 노드를 위해 열어두는 포트(동기화용).
 # 3888은 리더 선출을 위한 election 용 포트.
 # server. 뒤에 오는 숫자는 노드의 식별자가 된다. 

admin.enableServer=true
admin.serverPort=8000
admin.commandURL=/commands
 # 관리 서버 설정.
 # 사용하도록 설정하고 웹브라우저로 접속하면 커맨드 정보를 볼 수 있다.

[kafka@kafka ~]$  mkdir /home/kafka/zookeeper/datadir
[kafka@kafka ~]$  echo 1> /home/kafka/zookeeper/datadir/myid
 # myid에 적는 숫자는 zoo.cfg 에 기록한 서버 번호와 일치 해야 한다.

2020-08-312022-05-06

Zookeeper #1. 개요

Zookeeper #1. 개요
Zookeeper #2. 설치와 설정
Zookeeper #3. 구동과 확인

Kafka #1. 개요
Kafka #2. 설치

아파치 주키퍼(Apache ZooKeeper)는 아파치 소프트웨어 재단 프로젝트중의 한 소프트웨어 프로젝트로서 공개 분산형 구성 서비스, 동기 서비스 및 대용량 분산 시스템을 위한 네이밍 레지스트리를 제공한다. 주키퍼는 하둡의 한 하위 프로젝트이었으나 지금은 독립적인 상위 프로젝트이다. 주키퍼의 아키텍처는 중복 서비스를 이용한 고가용성을 제공한다. 클라이언트는 주키퍼 마스터가 응답을 하지 않으면 다른 주키퍼 마스터에게 요청을 한다. 주키퍼 노드들은 파일 시스템이나 trie 데이터구조와 비슷한 구조의 네임 스페이스안에 데이터들을 저장한다. 클라이언트들은 이 노드들에게서 읽거나 쓴다. ¹

쉬운 이해를 위해 설명하자면 ‘Hadoop Eco System’ 등의 고가용성 제공을 위한 도구이다. 즉, Hadoop 네임노드 이중화, Kafka 클러스터, NiFi 클러스터링 등을 위한 관리 기능을 제공한다.

기본적으로 1개 이상 ² 의 홀수개 노드(인스턴서,서버)들로 구성되며 주 용도는 다음과 같다.
– 설정 관리 : 클러스터의 설정 정보 관리
– 클러스터 관리 : 클러스터의 서버가 추가되거나 제외될 때 그 정보를 클러스터 노드 간 공유
– 리더 선출 : 흔한 이중화 개념에서 ‘액티브’ 노드로 사용할 노드를 선택
– 락 관리 및 동기화 서비스 : 클러스터 쓰기,연산 과다 등으로 인한 데이터 불일치를 최소화 하기 위한 락(Lock) 수행 및 동기화

이후 진행할 NiFi 클러스터 설정, Kafka, Hadoop 설정을 위한 최소한의 개념을 살펴보자면

① zookeeper 노드 간 정보를 확인하고 Leader가 선택된다.
② kafka broker들이 zookeeper 에 연결된다.
zookeeper는 kafka broker 3개 중 하나를 리더로 선출한다.
zookeeper 노드들은 kafka broker 정보들을 공유한다.
③ consumer는 메시지를 읽기 위해 zookeeper로부터 kafka broker 중 리더가 누구인지 확인한다.
④ consumer는 리더로 선출된 broker로부터 메시지를 받는다.
※ consumer, broker 등에 대한 이야기는 kafka 에 대해 이야기 할 때 설명하겠다.
위 과정은 실제 동작과는 차이가 있다. 단지 이해를 쉽게 하기 위해 서술하였다.

쉽게 설명하자면 노드가 여러개일 경우 노드들 중 실제 동작을 하게 될 노드를 선택하고 사용자가 통신하기 위한 노드의 정보를 관리하고 사용자에 전달해 주는 역할을 한다.
당연한 이야기지만 노드에 장애가 있을 경우 다음 노드를 지정해주는 역할도 한다.