Kafka 알아보고 설치해보기 - TIL230712

📚 KDT WEEK 15 DAY 3 TIL

• Kafka
• Kafka 아키텍처
• Kafka 기타 기능
• Kafka 설치 및 간단 실습

---

 

🟥 Kafka

 

 

실시간 데이터를 처리하기 위해 설계된 오픈소스 분산 스트리밍 플랫폼

• 분산 아키텍처를 따르기 때문에 Scale Out 형태로 스케일 가능
• Scalability와 Fault Tolerance를 제공하는 Publish-Subscription(Producer-Consumer) 메시징 시스템
  • 중간에 버퍼(큐)를 두고, Producer와 Consumer가 독립적으로 돌아가게 만든 시스템
• High Throughput, Low Latency로 실시간 데이터 처리에 맞게 구현
• 메시지를 정해진 보유기한(Retention period)동안 저장하게 된다

 

특이점

• 메시지 생산과 소비가 분리됨으로써 BackPressure(데이터 과생산) 방지
• ksqlDB를 통해 SQL로도 실시간 이벤트 데이터 처리 가능
• 데이터 복제 및 분산 커밋 로그 기능을 제공하여 장애 대응이 용이
• 브로커(서버)를 추가하여 쉽게 수평 확장 가능
• 커넥터와 통합 도구로 구성된 풍부한 에코시스템을 갖추고 있어 다른 데이터 시스템 및 프레임워크에 쉽게 연동 가능
  • ex) Kafka Connect, Kafka Schema Registry 등

 

! 분산 시스템

분산 시스템에서는 Fault Tolerance 등을 위해 데이터를 다수의 서버에 분산해서 저장함

100대 서버로 구성된 분산 시스템에 레코드를 하나 쓴다면 그 레코드를 바로 읽을 수 있는 가? > Consistency에 따라 다름

 

• Strong Consistency : 데이터를 쓸 때 모든 서버에 복제가 완료될 때까지 기다린 뒤, 레코드를 읽을 수 있게 됨
• Eventual Consistency : 데이터를 일부 서버에만 복제하고, 바로 레코드를 읽을 수 있도록 함

 

---

 

🟦 Kafka 아키텍처

 

1. Topic

Kafka에서는 Topic이라는 데이터 이벤트 스트림을 가지고 있다. 

Topic안에 데이터가 생성된 순서대로 쌓여 저장된다.

 

Producer(데이터 생산자)가 Topic을 만들고 Consumer(데이터 소비자)가 Topic에서 데이터를 읽는 구조이다.

이때 Consumer는 다수로 존재하여 Topic을 읽어들이는 것이 가능하다.

 

++) 여기서 중요한 점은 Consumer가 데이터를 읽는다고해서 데이터가 사라지지는 않는다는 것이다.

Topic은 앞서 Kafka 특징으로 소개했던 Retention period가 존재하여, 이 기간동안 데이터가 존재하게 된다.

따라서 Consumer별로 어느 위치의 데이터를 읽고 있는지 위치 정보(Offset)를 유지하게 된다.

 

이 Topic에 쌓이는 데이터를 Message(Event)라고 한다.

• Event는 최대 1Mb의 크기이며 Header, Key, Value, Timestamp를 포함한 구조를 가진다.
• Header에는 경량 메타 데이터 정보가 담기고
• Key에는 Topic 데이터를 Partitioning할 때 필요한 데이터가 포함된다.

 

1-1. Partitioning

 

하나의 Topic은 확장성을 위해 다수의 Partition으로 나뉘어 저장된다.

보통은 Key가 있다면 키의 Hashing 값을 Partition의 수로 나눈 나머지로 결정된다.

 

 

Topic의 Partition

 

하나의 Partition은 Fail-over를 위해 Replication Partiton(복제본)을 갖는다.

이때 각 Partition별로 Leader와 Follower로 구분되어 Leader를 통해 쓰기가 이뤄지고, Leader/Follower를 통해 읽게 된다.

Partition이 복제본을 갖는 이유는 데이터 유실을 막기 위한 것 뿐만아니라 다수의 Consumer가 병렬적으로 읽을 수 있도록 하기 위한 것도 있다.

 

++) Partition의 용도

• 수평적으로 Partition의 수를 늘려, Topic이 가질 수 있는 데이터의 크기 늘리기 > Load Balancing으로 이어짐
• 특정 키를 가지고 레코드를 나누는 경우 > Segmantic Partitioning

 

이러한 옵션들은 나중에 Kafka에서 Topic의 파라미터에 의해 결정된다.

 

이름, Partition의 수, 복제본의 수(원본 포함), Consistency Level, 데이터 보존 기한, 메시지 압축 방식.. 등

 

지금까지 설명했던 개념들이 여럿 보인다.

추가적인 파라미터는 다음의 링크를 참고하자.

https://kafka.apache.org/documentation/#topicconfigs

Apache Kafka

 

2. Broker

실제 데이터를 저장하는 서버

Kafka 클러스터는 기본적으로 다수의 Broker로 구성된다. (Broker = Server = Node)

Broker들이 실제로 Producer/Consumer들과 통신을 수행하게 되며, 부가 기능을 위한 다른 서비스들이 이곳에 추가된다.

또한 Topic의 Partition을 실제로 관리하며, 하나의 Broker에서 최대 4000개의 Partition을 처리하고 기록한다.

 

출처 : http://cloudurable.com/

 

이때 누가 어디에 Broker, Topic에 대한 정보를 관리할까?

 

3. ZooKeeper / KRaft

메타 정보 관리는 Apache ZooKeeper와 KRaft(Kafka Raft)을 보통 사용하게 된다.

• Broker 리스트 관리(누가 Controller(Broker 대장)인가?)
• Topic 리스트 관리(Partition 관리, Partition의 Replica 관리)
• Topic별 Access Control Lists) 관리 등

최근에는 ZooKeeper 관련 문제가 많아 KRaft로 많이 넘어오고 있다고 한다.

 

이제까지 설명한 Kafka의 아키텍처를 전체적인 그림으로 보면 다음과 같다.

 

출처: https://mail-narayank.medium.com/

 

Partition은 다시 여러 개의 Segment로 구성된다.

Segment는 물리적으로 디스크에 저장되는 로그 파일과 같다. (Immutable하고, Append만 가능한 형태)

Segment는 최대 크기가 존재해서, 이 크기를 넘어가면 새로운 Segment 파일을 만들어낸다.

데이터베이스에서 Data Recovery나 Replay에 사용하는 Commit Log라고 할 수 있다.

 

 

---

 

🟧 Kafka 기타 기능

 

1. Kafka Connect

Kafka 위에 만들어진 중앙집중 데이터 허브(별도의 서버 필요)

데이터 시스템들간의 데이터를 주고 받는 용도로 Kafka를 사용할 수 있게 만든다.

Kafka의 Prodcuer와 Consumer 역할을 Kafka Connect가 실행하게 되는 것이다.

 

Kafka Connect, 출처 : kakao tech

 

여기서 Source Task를 수행하는 것이 Producer의 역할, Sink Task를 수행하는 것이 Consumer의 역할을 한다.

 

Kafka 양쪽에 Kafka Connect를 두는 것으로, 특별한 코딩없이 간단한 환경설정으로 데이터를 읽어 Kafka 클러스터에 저장하거나 Kafka 클러스터로부터 읽어와서 S3와 같은 저장소에 저장할 수 있게 된다. (ETL)

 

2. Kafka Schema Registry

Kafka Topic에 들어온 데이터의 포맷이 어떤 형태이어야 하는지 기록되는 저장소 역할

따라서 Topic 메시지 데이터에 대한 스키마 관리 및 검증하는데 사용하게 된다.

 

Schema ID(와 버전)를 사용해서 Schema Evolution(포맷 변경)을 지원한다.

보통 압축 포맷은 AVRO를 사용한다.(스키마 임베드 가능!)

 

 

 

Kafka Schema Registry 구조

 

그림처럼 Schema Registry가 발급한 SchemaID를 통해 스키마 정보를 Consumer가 이해할 수 있게 된다.

이 과정에서 사용하는 Serialization과 Deserialization은 다음과 같이 이해하면 된다.

 

Serialization(직렬화)

• 객체를 저장하거나 전송할 수 있는 형태로 변환하는 프로세스
• 데이터를 일렬로 늘려서 압축하고 데이터의 스키마 정보를 추가하는 등의 작업

 

Deserialization(역직렬화)

• Serialized된 데이터를 다시 사용할 수 있는 형태로 변환하는 프로세스
• 데이터의 압축을 해제하거나 스키마 정보를 토대로 데이터 포맷 검증 수행 작업

 

직렬화와 역직렬화는 보통 Kafka 관련 라이브러리를 사용한다.

 

3. REST Proxy

Producer, Kafka Cluster, Consumer는 보통 같은 네트워크에서 사용하게 된다.

이때 외부에서 Kafka Cluster의 특정 Topic 데이터를 읽거나 쓸 때 발생하는 보안 문제를 해결하기 위한 방법이다.

 

클라이언트가 API 호출을 사용하여 Kafka를 사용 가능하게 하며,

Message Serialization과 Deserialization을 대신 수행하고 Load Balancing도 수행하게 된다.

Message를 생성 및 소비하고 Topic을 관리하는 간단하고 표준화된 방법을 제공하는 것이다.

 

4. Kafka Streams

Kafka Topic을 소비하고 생성하는 실시간 스트림 처리 라이브러리

 

Spark Streaming가 Kafka Topic을 처리하는 경우 micro batch에 가깝고,

Kafka Streams는 Topic의 데이터를 하나씩 처리하는 조금 더 Realtime에 가까운 방법이다.

 

5. ksqlDB

Kafka Streams로 구현된 스트림 처리 데이터베이스이다.(KSQL 대체)

SQL과 유사한 쿼리 언어를 지원하고, 조금 특별하게 연속 쿼리라는 기능도 지원한다.

연속 쿼리는 SELECT와 같은 쿼리를 넣었을 때, 데이터가 실시간으로 도착할 때마다 업데이트되는 기능이다.

 

---

 

🟩 Kafka 설치 및 간단 실습

 

Kafka docker-compose 파일을 통해 설치한다.

 

Conduktor.io 회사에서 만든 Kafka 레포를 참고하자.

https://github.com/conduktor/kafka-stack-docker-compose

GitHub - conduktor/kafka-stack-docker-compose: docker compose files to create a fully working kafka stack

 

다양한 yml 파일 중 여러 가지 기능을 담고있는 Full Stack 버전을 설치한다.

이 중에는 conduktor가 제공하는 웹 UI도 있다.

 

git pull https://github.com/conduktor/kafka-stack-docker-compose.git
docker compose -f full-stack.yml up

 

두 번째 명령어를 수행하는 데에는 대략 5분 정도 기다려야 한다.

 

작업이 모두 끝나면 conduktor의 웹 UI에 접속해보자.

 http://localhost:8080/ 로 접속할 수 있고, 아이디와 비밀번호는 각각 admin@admin.io , admin 이다.

 

웹 UI 화면

 

왼쪽 메뉴에 보이는 Topics, Schema Registry, Consumer Groups 등 이제껏 알아봤던 기능들을 쉽게 확인할 수 있다.

참고로 로컬에 설치하여 운용하는 테스트용 Kafka로써 활용되는 환경이므로, Brokers도 하나만 존재한다.

 

이제 python을 활용한 Kafka를 사용하기 위해 관련 환경을 설치해주자.

pip3 install kafka-python

 

그리고 producer.py를 작성한다.

 

from time import sleep
from json import dumps
from kafka import KafkaProducer

# producer 생성
producer = KafkaProducer(
 bootstrap_servers=['localhost:9092'], # Broker에 해당하는 서버 입력
 value_serializer=lambda x: dumps(x).encode('utf-8') # dumps는 dictinary를 json string으로 바꿔줌
)

# 0.5초마다 데이터 넣기
for j in range(50):
 print("Iteration", j)
 data = {'counter': j}
 producer.send('topic_test', value=data) # 이름이 topic_test인 Topic에 해당 데이터 전송
 sleep(0.5)

 

이 파이썬 파일을 아까 git clone된 폴더에 넣고 실행시켜보자.

 

 

Kafka가 실행되어 있지 않다면 "NoBrokersAvailable"이란 에러 메시지가 발생한다.

그리고 웹 UI로 결과를 확인할 수 있다.

 

topic_test라는 topic이 생겼다.

 

topic_test 내부 모습

 

별다른 Key나 옵션을 지정해주지 않아 Key는 null로 기록되어있다.

이렇게 Key가 Null인데, 여러 개의 Partition을 가지고 있다면 라운드 로빈형식으로 분산하게 된다.

 

이번에는 consumer.py를 작성해보자.

 

from kafka import KafkaConsumer
from json import loads
from time import sleep

# consumer 생성
consumer = KafkaConsumer(
 'topic_test', # topic_test Topic에서 데이터를 읽어온다.
 bootstrap_servers=['localhost:9092'], # Broker 서버 입력
 auto_offset_reset='earliest', # 가장 먼저 들어온 데이터부터 읽기, 반대는 'latest'
 enable_auto_commit=True, # False인 경우 commit 함수로 명시적으로 offset 위치를 커밋해야함(안전)
 group_id='my-group-id', # Topic을 같이 소비할 Conumser 그룹
 value_deserializer=lambda x: loads(x.decode('utf-8')) # 아까 producer와 반대 작업
)

# 읽어온 데이터를 출력
for event in consumer:
 event_data = event.value
 # Do whatever you want
 print(event_data)
 sleep(2)

 

 

consumer도 정상적으로 데이터를 잘 읽어온다.