[Study] Kafka - nrbae/happy-project GitHub Wiki

참고 서적 : 카프카, 데이터 플랫폼의 최강자

1. 카프카와 주키퍼

1-1. 주키퍼

주키퍼는 분산 애플리케이션을 위한 코디네이션 시스템. 분산되어있는 각 애플리케이션의 정보를 중앙에 집중하고 구성 관리, 그룹 관리 네이밍, 동기화 등의 서비스를 제공.

서버 여러대를 앙상블(클러스터)로 구성하고, 분산 애플리케이션들이 각각 클라이언트가 되어 주키퍼 서버들과 커넥션을 맺은 후 상태 정보등을 주고받음. 클러스터로 구성된 서버중 한대를 리더로 선정.
상태 정보들은 주키퍼의 지노드(znode)라 불리는 곳에 키-값(key-value) 형태로 저장됨.
신뢰성 있는 서비스를 위해 클러스터라는 호스트 세트를 구성(ex. 3대). 클러스터로 구성된 주키퍼는 살아있는 노드 수가 과반수 이상 유지되면 지속적인 서비스가 가능(홀수로 서버를 구성).

1-2. 카프카

대용량, 대규모 메시지 데이터를 빠르게 처리하도록 개발된 메시징 플랫폼.

참고 : Kafka 운영자가 말하는 처음 접하는 Kafka

특징
- 프로듀서와 컨슈머의 분리 : 프로듀서는 카프카로 메시지를 보내는 역할만, 컨슈머는 카프카에 저장된 메시지를 가져오는 역할만으로 분리.
- 멀티 프로듀서, 멀티 컨슈머 : 하나 이상의 토픽에 메시지를 보내거나 가져올 수 있음.
- 디스크에 메시지 저장
- 클러스터 구성의 확장성
- 분산처리, 배치처리 등의 기법을 통한 고성능
카프카 클러스터는 홀수 운영 구성할 필요 없음.
주키퍼와 카프카는 동일한 서버가 아닌 별도의 주키퍼 서버와 별도의 카프카 서버로 구성하는것을 권장.

2. 설치

2-1. 주키퍼 설치 (3 instance in 1 node)

참고 : How to setup and run multiple zookeeper instances on one box

패키지 설치 (자바 설치 필수)

$ wget https://archive.apache.org/dist/zookeeper/zookeeper-3.4.13/zookeeper-3.4.13.tar.gz
$ tar zxf zookeeper-3.4.13.tar.gz
$ ln -sf /home1/irteam/install/zookeeper-3.4.13 /home1/irteam/apps/zookeeper1

3 instance를 위해 패키지 디렉토리 3개 생성

$ ls -al apps/
lrwxrwxrwx  1 irteam irteam   38 Nov 11 18:04 zookeeper1 -> /home1/irteam/install/zookeeper-3.4.13
lrwxrwxrwx  1 irteam irteam   40 Nov 11 19:00 zookeeper2 -> /home1/irteam/install/zookeeper-3.4.13_2
lrwxrwxrwx  1 irteam irteam   40 Nov 11 19:00 zookeeper3 -> /home1/irteam/install/zookeeper-3.4.13_3

데이터 디렉토리 생성 (지노드의 복사본인 스냅샷과 트랜잭션 로그들이 저장됨)

$ mkdir -p zk_data1
$ echo > 1 zk_data1/myid

$ mkdir -p zk_data2
$ echo > 2 zk_data2/myid

$ mkdir -p zk_data3
$ echo > 3 zk_data3/myid

설정 파일 수정

$ cp /home1/irteam/apps/zookeeper1/conf/zoo_sample.cfg zoo.cfg
$ vi /home1/irteam/apps/zookeeper1/conf/zoo.cfg

[/home1/irteam/apps/zookeeper1/conf/zoo.cfg]
dataDir=/home1/irteam/zk_data1
clientPort=2181
server.1=localhost:2888:3888
server.2=localhost:2889:3889
server.3=localhost:2890:3890

[/home1/irteam/apps/zookeeper2/conf/zoo.cfg]
dataDir=/home1/irteam/zk_data2
clientPort=2182
server.1=localhost:2888:3888
server.2=localhost:2889:3889
server.3=localhost:2890:3890

[/home1/irteam/apps/zookeeper3/conf/zoo.cfg]
dataDir=/home1/irteam/zk_data3
clientPort=2183
server.1=localhost:2888:3888
server.2=localhost:2889:3889
server.3=localhost:2890:3890

주키퍼 실행/종료 및 클라이언트 접속

$ /home1/irteam/apps/zookeeper1/bin/zkServer.sh [start|stop]

$ /home1/irteam/apps/zookeeper1/bin/zkCli.sh -server 127.0.0.1:2181
$ /home1/irteam/apps/zookeeper1/bin/zkCli.sh -server 127.0.0.1:2182
$ /home1/irteam/apps/zookeeper1/bin/zkCli.sh -server 127.0.0.1:2183

2-2. 카프카 설치 (3 instance in 1 node)

패키지 설치

$ wget http://apache.tt.co.kr/kafka/2.0.0/kafka_2.11-2.0.0.tgz
$ tar xzf kafka_2.11-2.0.0.tgz
$ ln -sf /home1/irteam/install/kafka_2.11-2.0.0.tgz /home1/irteam/apps/kafka

설정 파일 수정

참고

Running a Multi-Broker Apache Kafka 0.8 Cluster on a Single Node

broker configs

$ cp /home1/irteam/apps/kafka/config/server.properties /home1/irteam/apps/kafka/config/server1.properties
$ cp /home1/irteam/apps/kafka/config/server.properties /home1/irteam/apps/kafka/config/server2.properties
$ cp /home1/irteam/apps/kafka/config/server.properties /home1/irteam/apps/kafka/config/server3.properties

$ vi /home1/irteam/apps/kafka/config/server1.properties

[/home1/irteam/apps/kafka/config/server1.properties]
broker.id=1
port=9092
log.dirs=/home1/irteam/kafka_data1
zookeeper.connect=localhost:2181,localhost:2182,localhost:2183/test-kafka
delete.topic.enable=true
auto.create.topics.enable=true

[/home1/irteam/apps/kafka/config/server2.properties]
broker.id=2
port=9093
log.dirs=/home1/irteam/kafka_data2
zookeeper.connect=localhost:2181,localhost:2182,localhost:2183/test-kafka
delete.topic.enable=true
auto.create.topics.enable=true

[/home1/irteam/apps/kafka/config/server3.properties]
broker.id=3
port=9094
log.dirs=/home1/irteam/kafka_data3
zookeeper.connect=localhost:2181,localhost:2182,localhost:2183/test-kafka
delete.topic.enable=true
auto.create.topics.enable=true

카프카 실행/종료

$ env JMX_PORT=9999 /home1/irteam/apps/kafka/bin/kafka-server-start.sh -daemon /home1/irteam/apps/kafka/config/server1.properties
$ env JMX_PORT=10000 /home1/irteam/apps/kafka/bin/kafka-server-start.sh -daemon /home1/irteam/apps/kafka/config/server2.properties
$ env JMX_PORT=10001 /home1/irteam/apps/kafka/bin/kafka-server-start.sh -daemon /home1/irteam/apps/kafka/config/server3.properties

$ /home1/irteam/apps/kafka/bin/kafka-server-stop.sh

주키퍼에서 카프카 정보 확인

$ /home1/irteam/apps/zookeeper1/bin/zkCli.sh -server 127.0.0.1:218[1-3]

[zk: 127.0.0.1:2183(CONNECTED) 0] ls /test-kafka/brokers/ids
[1, 2, 3]

카프카 로그

/home1/irteam/apps/kafka/logs/server.log

기본 동작 테스트

# 토픽 생성
$ /home1/irteam/apps/kafka/bin/kafka-topics.sh --create --zookeeper localhost:2181,localhost:2182,localhost:2183/test-kafka --replication-factor 1 --partitions 1 --topic test-topic

# 메시지 퍼블리싱
$ /home1/irteam/apps/kafka/binfka-console-producer.sh --broker-list localhost:9092,localhost:9093,localhost:9094 --topic test-topic
>This is a first message.
>This is a second message.

# 메시지 가져오기
$ /home1/irteam/apps/kafka/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092,localhost:9093,localhost:9094 --topic test-topic --from-beginning
This is a first message.
This is a second message.

# 토픽 삭제
$ /home1/irteam/apps/kafka/bin/kafka-topics.sh --zookeeper localhost:2181,localhost:2182,localhost:2183/test-kafka --topic test-topic --delete

3. 카프카 디자인

분산 시스템(for 높은 처리량, 빠른 메시지 전송, 운영 효율화), 페이지 캐시, 배치 전송 처리 등의 기능 구현

대용량 실시간 데이터 처리 : 배치 전송, 파티션, 분산 기능
데이터 안정적 저장 : 리플리케이션 기능, 분산된 서버에서 자동으로 파티션의 리더를 선출하는 기능

3-1. 카프카 디자인 특징

분산 시스템

분산 시스템 : 같은 역할을 하는 여러 대의 서버로 이뤄진 서버 그룹.
단일 시스템보다 높은 성능, 하나의 서버 또는 노드 등이 장애 발생시 다른 서버 또는 노드가 대신 처리 가능, 시스템 확장이 용이하다는 장점이 있음.

페이지 캐시

페이지 캐시 : OS의 전체적인 성능 향상을 위해 물리적 메모리에 애플리케이션이 사용하는 부분을 할당하고 남은 잔여 메모리의 일부로 페이지 캐시를 유지. 디스크에 읽고 쓰지 않고 페이지 캐시를 통해 읽고 쓰면 처리 속도가 빠름.
카프카는 빠른 엑세스를 위해 OS의 페이지 캐시를 이용.

배치 전송 처리

빈번하게 일어나는 작은 I/O들이 속도를 저하시키는 원인이 될 수 있음. 카프카에서는 작은 I/O들을 묶어 처리할 수 있도록 배치 작업으로 처리(작은 메시지들을 묶어서 한번에 보냄).

3-2. 카프카 데이터 모델

토픽 : 메시지를 받을 수 있도록 논리적으로 묶은 개념
파티션 : 토픽을 구성하는 데이터 저장소. 수평 확장이 가능한 단위임.

토픽

카프카 클러스터는 토픽이라 불리는 곳에 데이터를 저장. 데이터를 구분하기 위한 단위. 프로듀서들은 원하는 토픽으로만 메시지를 보내고 컨슈머는 원하는 토픽의 메시지만 가져올 수 있음.

(ex) sbs-news / sbs-video / kbs-news / kbs-video

파티션

카프카에서 파티션이란 토픽을 분할한 것. 메시징 큐 시스템의 경우 메시지의 순서가 보장되어야 하기 때문에 프로듀서 4개가 파티션 1개로 메시지를 보내더라도 이전 메시지 처리가 완료된 후 다음 메시지를 처리하게 됨. 빠른 전송을 위해서는 토픽의 파티션을 늘려줘야하며, 그 수만큼 프로듀서 수도 늘려야 제대로 된 효과를 볼 수 있음.

파티션 수를 늘릴때의 주의점
- 파일 핸들러의 낭비 : 각 파티션은 브로커의 디렉토리와 매핑되고, 저장되는 데이터마가 2개의 파일(인덱스와 실제 데이터)이 있음. 파티션 수가 많을수록 파일 핸들 수가 많아져 리소스를 낭비.
- 장애 복구 시간 증가 : 카프카는 높은 가용성을 위해 리플리케이션을 지원. 각 파티션마다 리플리케이션이 동작하게 되며, 하나는 파티션의 리더이고 나머지는 파티션의 팔로워가 됨. 리더를 선출하는 역할은 컨트롤러로 지정된 브로커가 수행.
- 카프카에서 파티션 수의 증가는 아무때나 가능하지만, 파티션의 수를 줄이는 방법은 없음. 파티션 수를 줄이고 싶으면 토픽을 삭제해야함.

오프셋과 메시지 순서

파티션마다 메시지가 저장되는 위치를 오프셋이라고 함. 오프셋은 파티션 내에서 유일하고 순차적으로 증가하는 숫자(64비트 정수) 형태. 오프셋은 하나의 파티션 내에서만 유일한 숫자임(파티션마다 유니크한 값을 가짐). 카프카에서는 이 오프셋을 이용해 메시지의 순서를 보장함. 오프셋 순서대로만 데이터를 가져갈 수 있음.

3-3. 카프카의 고가용성과 리플리케이션

카프카의 리플리케이션은 토픽 자체를 리플리케이션하는 것이 아니라, 토픽을 이루는 각각의 파티션을 리플리케이션 함.

리플리케이션 팩터와 리더, 팔로워의 역할

리플리케이션 팩터
- replication factor 수를 조정하여, replication의 수를 몇 개로 할지 관리자가 조정.
- /usr/local/kafka/config/server.properties 설정 파일에서 default.replication.factor 수정을 통해 리플리케이션 팩터 값을 변경할 수 있음(기본값 : 1).
- 이 설정 값은 아무런 옵션을 주지 않고 토픽을 생성할때 적용됨.
- 각 토픽별로 다른 리플리케이션 팩터 값을 설정할 수 있음. 토픽에 저장되는 데이터의 중요도에 따라 리플리케이션 팩터를 설정. (2 or 3)
- 클러스터 내 모든 브로커에 동일하게 설정해야하며, 변경 후 브로커 1대씩 재시작해야 변경 내용이 적용됨.
- 리플리케이션 팩터 값이 커질수록 카프카 클러스터 내 필요한 저장소 크기가 필요하며, 브로커의 리소스 사용량이 증가할 수 있음.
- (ex) 토픽 생성 / 확인

$ /usr/local/kafka/bin/kafka-topics.sh \ --zookeeper zk-server1:2181,zk-server2:2181,zk-server3:2181/test-node \ --topic test-topic --partitions 1 --replication-factor 2 --create
$ /usr/local/kafka/bin/kafka-topics.sh \ --zookeeper zk-server1:2181,zk-server2:2181,zk-server3:2181/test-node \ --topic test-topic --describe

리더(원본)와 팔로워(복제본)
- 모든 읽기와 쓰기는 리더를 통해서만 일어남. 팔로워는 리더의 데이터를 그대로 리플리케이션만 함.
- 리더와 팔로워는 저장된 데이터의 순서도 일치하고 동일한 오프셋과 메시지들을 갖게 됨.

리더와 팔로워의 관리

ISR(In Sync Replica) : 현재 리플리케이션되고 있는 리플리케이션 그룹. 리더와 팔로워간 데이터 불일치 현상을 방지하기 위해 도입된 개념. ISR에 속해 있는 구성원만이 리더의 자격을 가진다는 규칙이 있음.
리더는 팔로워들이 주기적으로 데이터를 확인하고 가져가고 있는지를 확인하여 일정 주기(replica.lag.time.max.ms)만큼 확인 요청이 오지 않는다면 해당 팔로워를 ISR 그룹에서 추방시킴.

모든 브로커가 다운된 상황

마지막 리더가 살아나기를 기다린다.
- 메시지 손실은 없지만 정상화까지 시간이 오래걸릴 수 있음.
- /usr/local/kafka/config/server.properties에서 unclean.leader.election.enable = false
ISR에서 추방되었지만 먼저 살아나면 자동으로 리더가 된다.
- 메시지 손실은 있겠지만 빠른 서비스를 제공할 수 있음.
- /usr/local/kafka/config/server.properties에서 unclean.leader.election.enable = true

카프카에서 사용하는 주키퍼 지노드

--/test-node
      |
       -----/controller
      |
       -----/brokers
      |       |
      |        -----/ids (임시 노드)
      |       |
      |        -----/topics
      |               |
      |                -----/test-topic
      |
       -----/consumers
      |       |
      |        -----/test-consumer
      |               |
      |                -----/offsets (영구 노드)
      |
       -----/config
              |
               -----/topics
                      |
                       -----/test-topic

/test-node/controller
- 현재 카프카 클러스터의 컨트롤러 정보.
- 카프카에서는 클러스터 내 브로커 중 하나를 컨트롤러로 선정. 컨트롤러는 브로커 레벨에서의 실패를 감지하고 실패한 브로커에 의해 영향받는 모든 파티션의 리더 변경을 책임짐.
/test-node/brokers
- 브로커 관련 정보들.
- ids : 브로커는 시작시 /brokers/ids에 broker.id로 지노드를 작성해 자신을 등록. 주키퍼의 임시노드를 사용. 브로커가 종료되거나 다운되면 지노드는 사라짐.
- topics : 클러스터 내 토픽 정보들(토픽의 파티션 수, ISR 구성 정보, 리더 정보) 등을 확인.
/test-node/consumers
- 컨슈머 관련된 정보들.
- offsets : 컨슈머가 각각의 파티션들에 대해 어디까지 읽었는지를 기록하는 오프셋 정보.
- 카프카 0.9 버전 이후는 컨슈머 오프셋 저장 장소를 카프카의 토픽(__consumer_offsets)으로 하길 권장.
/test-node/config
- 토픽의 상세 설정 정보.