[Kafka] Kafka 설치 및 Kafka Python 프로그래밍 (Docker Compose)

⭐ Kafaka 설치

• Github reop 사용

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

• 컴퓨터 사양이 좋지 않다면 zk-single-kafka-single.yml 또는 zk-single-kafka-multiple.yml 사용

docker compose -f full-stack.yml up

• 로컬 호스트 8080 포트로 접속하면 Conduktor 웹페이지로 연결
  

⭐ Kafka Python 프로그래밍

1. Kafka 프로그래밍 옵션

• Java

  • Apache Kafka Java Client: 아파치 카프카의 공식 Java 클라이언트 라이브러리
  • Spring Kafka: 스프링 프레임워크와 Kafka를 통합하기 위한 라이브러리

• Python

  • Confluent Kafka Python: Confluent에서 개발한 공식 Kafka Python 클라이언트 라이브러리
  • Kafka-Python: 또다른 파이썬 기반 라이브러리

• .NET

  • Confluent Kafka .NET Client: Confluent에서 개발한 공식 Kafka .NET 클라이언트 라이브러리

• Go

  • Sarama: Go 언어용 Kafka 클라이언트 라이브러리로

• Node.js

  • node-rdkafka: librdkafka를 기반으로 한 Node.js용 Kafka 클라이언트 라이브러리
  • kafka-node: Node.js용 Kafka 클라이언트 라이브러리

2. Python 모듈 설치

• pip3 install kafka-python
  

3. Producer

• 간단한 Producer 만들기

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

producer = KafkaProducer(
   bootstrap_servers=['localhost:9092'],
   value_serializer=lambda x: dumps(x).encode('utf-8')
)

for j in range(999):
   print("Iteration", j)
   data = {'counter': j}
   producer.send('topic_test', value=data)
   sleep(0.5)

• 간단한 Producer 실행 (998번까지 실행됨)

python3 producer.py

• Kafka 웹 콘솔에서 Topic 생성 여부 확인
  

4. Consumer

• 간단한 Consumer 만들기

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

consumer = KafkaConsumer(
   'topic_test',
   bootstrap_servers=['localhost:9092'],
   auto_offset_reset='earliest',
   enable_auto_commit=True,
   group_id='my-group-id',
   value_deserializer=lambda x: loads(x.decode('utf-8'))
)
for event in consumer:
   event_data = event.value
   # Do whatever you want
   print(event_data)
   sleep(2)

• 간단한 Consumer 실행

python3 consumer.py

5. 웹 콘솔에서 Consume 확인