데이터로 보는 세상

데이터로 보는 세상에서는 실시간 데이터 스트리밍의 중요성이 갈수록 커지고 있습니다. 오늘은 Apache Kafka와 Python을 연동하여 실시간 데이터 스트리밍 문제를 해결하는 방법에 대해 알아보겠습니다. Kafka는 대규모 데이터 스트리밍을 효율적으로 처리할 수 있는 분산 메시지 시스템으로, 다양한 언어와의 연동이 가능합니다. Python은 데이터 분석 및 처리에 강력한 라이브러리를 제공하여 Kafka와의 통합에 매우 유용합니다.

Kafka와 Python의 설치

먼저, Kafka와 관련된 환경을 설정해야 합니다. Kafka를 설치하기 위해 공식 Apache Kafka 다운로드 페이지에서 최신 버전을 다운로드한 후 압축을 해제합니다. Kafka는 Zookeeper에 의존하기 때문에 Zookeeper도 함께 설정해야 합니다.

Kafka 및 Zookeeper 실행

bin/zookeeper-server-start.sh config/zookeeper.properties

# Kafka 브로커 실행
bin/kafka-server-start.sh config/server.properties

이 명령어를 통해 Kafka와 Zookeeper의 서버를 실행할 수 있습니다. 이제 Kafka가 준비되었습니다.

Python 환경 설정

Python과 Kafka를 연동하기 위해 confluent-kafka 패키지를 설치합니다. 이 패키지는 Kafka 프로듀서와 소비자를 쉽게 사용할 수 있도록 도와줍니다.

pip install confluent-kafka

Kafka 프로듀서 예제

이제 Kafka에 메시지를 보내는 간단한 프로듀서를 만들어 보겠습니다. 아래 코드는 "test_topic"이라는 주제로 메시지를 전송하는 예제입니다.

from confluent_kafka import Producer
import socket

# 프로듀서 설정
conf = {
    'bootstrap.servers': 'localhost:9092',
    'client.id': socket.gethostname()
}

producer = Producer(**conf)

def delivery_report(err, msg):
    if err is not None:
        print(f'Message delivery failed: {err}')
    else:
        print(f'Message delivered to {msg.topic()} [{msg.partition()}]')

# 메시지 전송
for i in range(10):
    producer.produce('test_topic', key=str(i), value=f'Hello Kafka {i}', callback=delivery_report)

# 큐에 남아 있는 모든 메시지를 전송
producer.flush()

위 코드를 실행하면 "test_topic"에 0부터 9까지의 메시지가 전송됩니다. delivery_report 함수는 메시지가 성공적으로 전송되었는지 여부를 확인합니다.

Kafka 소비자 예제

이제 Kafka에서 메시지를 소비하는 소비자를 만들어 보겠습니다. 아래 코드는 "test_topic"에서 메시지를 수신하는 예제입니다.

from confluent_kafka import Consumer, KafkaError

# 소비자 설정
conf = {
    'bootstrap.servers': 'localhost:9092',
    'group.id': 'my_group',
    'auto.offset.reset': 'earliest'
}

consumer = Consumer(**conf)
consumer.subscribe(['test_topic'])

# 메시지 소비
try:
    while True:
        msg = consumer.poll(1.0)  # 1초 대기
        if msg is None:
            continue
        if msg.error():
            if msg.error().code() == KafkaError._PARTITION_EOF:
                continue
            else:
                print(f'Consumer error: {msg.error()}')
                break
        print(f'Received message: {msg.value().decode("utf-8")}')
except KeyboardInterrupt:
    pass
finally:
    consumer.close()

위 코드를 실행하면 "test_topic"으로부터 메시지를 수신하여 출력합니다. consumer.poll(1.0) 함수는 1초 동안 메시지를 기다리며, 새로운 메시지가 도착하면 이를 수신합니다.

실시간 데이터 스트리밍의 활용

Kafka와 Python을 활용한 실시간 데이터 스트리밍은 다양한 사용 사례가 있습니다. 예를 들어, 웹 애플리케이션에서 사용자 행동 데이터를 수집하여 실시간으로 분석하거나, IoT 기기에서 발생하는 데이터를 수집하여 실시간으로 모니터링할 수 있습니다. Kafka의 높은 처리량과 안정성을 통해 대량의 데이터 스트림을 효과적으로 관리할 수 있습니다.

마무리

이 글에서는 Kafka와 Python을 연동하여 실시간 데이터 스트리밍 문제를 해결하는 기본적인 예제를 살펴보았습니다. Kafka의 강력한 기능과 Python의 유연성을 결합하면, 다양한 데이터 처리 및 분석 문제를 해결할 수 있는 가능성이 열립니다. 실시간 데이터를 효과적으로 처리하고 분석하여 더욱 풍부한 인사이트를 얻는 데 이 방법이 도움이 되길 바랍니다.

Zookeeper 실행

Kafka와 Python의 설치

Kafka 프로듀서 예제

Kafka 소비자 예제

실시간 데이터 스트리밍의 활용

마무리