Kafka - 이론, 실습(Ubuntu)

💡 준비물 : ubuntu 20.04.3, Kafka

설명

Kafka란 ?

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

특징

• pub/sub 메시지 시스템 : 생산자가 원하는 각 메시지를 게시할 수 있는 메시지 배포 패턴
• 실시간 데이터 피드 관리용도 플랫폼
  • 높은 처리량
  • 낮은 지연시간
• 분산 트랜잭션 로그로 구성된, 확장가능한 pub/sub 메시지 큐
• 주키퍼와 같이 동작
  • 구성 정보를 유지 관리하고, 이름을 지정하고, 분산 동기화를 제공하고, 그룹 서비스를 제공하기 위한 중앙 집중식 서비스
  • 분산형 코디네이션 서비스
  • 출처
    • https://zookeeper.apache.org/
• 사용 용도
  • 스트리밍 데이터 처리
• 기능
  • 분산 이벤트 스트리밍 플랫폼
  • 높은 처리량
  • 실시간 솔루션
  • 영구 메시징
• 특징
  • 메시지 대기열에 의존하지 않고 메시지를 로그에 추가하고 그대로 두며, 소비자가 메시지를 읽거나 보존 제한에 도달할 때까지 그대로 유지.
    • disk 메시지 저장 - offset
  • 풀 기반 접근 방식
    • 메시지 일괄처리 가능

설치

# 설치및 압축 해제
wget <https://downloads.apache.org/kafka/3.3.1/kafka_2.12-3.3.1.tgz>
tar -xzf kafka_2.12-3.3.1.tgz

실행

# zookeeper / kafka 실행
cd kafka_2.12-3.3.1/bin
./zookeeper-server-start.sh ../config/zookeeper.properties
./kafka-server-start.sh ../config/server.properties

topic 생성

# 토픽 생성
# 토픽이름 : test
# replication-factor : 복제 대상 파티션 개수 1개 (없음)
# partitions : 파티션 개수 3개
./kafka-topics.sh --create --bootstrap-server localhost:9092 --replication-factor 1 --partitions 3 --topic test

# 로그확인 가능
cat ../logs/controller.log

# 토픽 생성 확인
./kafka-topics.sh --describe --bootstrap-server localhost:9092

메시지 작성

# 메시지 작성
./kafka-console-producer.sh --bootstrap-server localhost:9092 --topic test
> test
> kafka
> start
> end

# 메시지 구독 
./kafka-console-consumer.sh --bootstrap-server localhost:9092 --from-beginning --topic test
test
kafka
start
end

종료

# KAFKA 환경 종료
# 카프카를 종료하려면 콘솔에서 Ctrl-C 로 아래 순서대로 종료해 주시면됩니다.

# 생산자 및 소비자 클라이언트 중지
# Kafka 브로커를 중지
# ZooKeeper 서버 중지

Kafka 간단 실습

Consumer Group

• 그룹에 속하는 컨슈머가 여러개면 로드밸런싱을 통해 자동으로 메시지를 분배.

# 그룹 생성
./kafka-topics.sh --create --bootstrap-server localhost:9092 --replication-factor 1 --partitions 3 --topic test1

# 그룹 확인
./kafka-console-consumer.sh --bootstrap-server localhost:9092 -group testgroup1  --from-beginning --topic test1
GROUP           TOPIC           PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG             CONSUMER-ID                                           HOST            CLIENT-ID
testgroup1       test1           0          0               0               0               console-consumer-eb58c4ca-fb34-45c7-867f-2ef0ad8951dd /127.0.0.1      console-consumer
testgroup1       test1           1          0               0               0               console-consumer-eb58c4ca-fb34-45c7-867f-2ef0ad8951dd /127.0.0.1      console-consumer
testgroup1       test1           2          0               0               0               console-consumer-eb58c4ca-fb34-45c7-867f-2ef0ad8951dd /127.0.0.1      console-consumer

• 컨슈머 그룹 테스트

# 메시지 입력
./kafka-console-producer.sh --bootstrap-server localhost:9092 --topic test1
> 1
> 2
> 3
> 4
> 5
> 6
> 7
> 8
> 9

# 그룹 확인
./kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group testgroup1

GROUP           TOPIC           PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG             CONSUMER-ID                                           HOST            CLIENT-ID
testgroup1      test1           0          3               3               0               console-consumer-672ffa72-f17e-43b0-b4be-c9304803391b /127.0.0.1      console-consumer
testgroup1      test1           1          3               3               0               console-consumer-672ffa72-f17e-43b0-b4be-c9304803391b /127.0.0.1      console-consumer
testgroup1      test1           2          3               3               0               console-consumer-672ffa72-f17e-43b0-b4be-c9304803391b /127.0.0.1      console-consumer

Python 간단 테스트

# kafka 패키지 설치
pip install kafka-python

Producer 옵션

Python-Producer 작성

# producer.py 작성
from kafka import KafkaProducer
from json import dumps
import time

# producer 객체 생성
producer = KafkaProducer(acks=0, compression_type='gzip', bootstrap_servers=['localhost:9092'],
                                 value_serializer=lambda x: dumps(x).encode('utf-8'))

start = time.time()
for i in range(10000):
    data = {'str' : 'result'+str(i)}
    producer.send('test1', value=data)
    producer.flush()
    print("elapsed :", time.time() - start)

# 실행
python producer.py
..
elapsed : 57.16879749298096
elapsed : 57.171818017959595
elapsed : 57.175421476364136

Consumer 옵션

Python-Consumer 작성

# consumer.py 작성
from kafka import KafkaConsumer
from json import loads
import time

# topic, broker list
consumer = KafkaConsumer(
        'test1',
        bootstrap_servers=['localhost:9092'],
        auto_offset_reset='earliest',
        enable_auto_commit=True,
        group_id='testgroup1',
        value_deserializer=lambda x: loads(x.decode('utf-8')),
        consumer_timeout_ms=1000
    )

# consumer list를 가져온다
print('[begin] get consumer list')
while True:
    for message in consumer:
        print("Topic: %s, Partition: %d, Offset: %d, Key: %s, Value: %s" % (
                    message.topic, message.partition, message.offset, message.key, message.value
                        ))
        #time.sleep(1)

    print('[end] get consumer list')

# 출력
..
Topic: test, Partition: 0, Offset: 96, Key: None, Value: {'str': 'result33'}
Topic: test, Partition: 0, Offset: 97, Key: None, Value: {'str': 'result34'}
Topic: test, Partition: 0, Offset: 98, Key: None, Value: {'str': 'result35'}
Topic: test, Partition: 0, Offset: 99, Key: None, Value: {'str': 'result36'}

Kafka Docker + UI

- https://devocean.sk.com/blog/techBoardDetail.do?ID=163980

Kafka-UI Tool 을 이용하여 Kafka 관리하기

출처 : https://needjarvis.tistory.com/607,
https://github.com/dpkp/kafka-python

GitHub - dpkp/kafka-python: Python client for Apache Kafka

[카프카] Python으로 Kafka에 전송(Producer)하고 가져오기(consumer)