Source: Dev.to

Introduction

은행이 실시간으로 사기를 감지하고 차단하는 방법이 궁금했던 적 있나요? 바로 이런 방식으로 이루어집니다.
이벤트 스트리밍은 데이터베이스, 센서, 클라우드 서비스, 소프트웨어 애플리케이션 등 다양한 소스에서 실시간으로 데이터를 캡처해 이벤트 스트림 형태로 저장하고, 필요에 따라 해당 스트림을 다양한 목적지 기술로 라우팅하는 실천 방법입니다.
Kafka는 서버와 클라이언트가 TCP 기반 고성능 바이너리 프로토콜을 통해 통신하는 분산 이벤트 스트리밍 플랫폼입니다.

Kafka를 구성하는 주요 요소는 다음과 같습니다.

1. Servers (brokers) – 데이터를 저장하고 검색합니다.
2. Clients (producers and consumers) – 이벤트를 발행하고 읽습니다.
3. Producers
4. Consumers
5. Topics and Partitions
6. Zookeeper (신버전에서는 Kraft)

이제 OpenWeather 웹사이트의 API를 이용해 날씨 데이터를 가져오고, Kafka producer로 토픽에 기록한 뒤, Kafka consumer로 토픽을 읽는 예제 프로젝트를 진행해 보겠습니다.

파이썬으로 Producer 만들기

from kafka import KafkaProducer
import requests
from dotenv import load_dotenv
import os
import json

load_dotenv()

API_KEY = os.getenv('API_KEY')

위 코드는 Producer가 동작하는 데 필요한 의존성을 불러옵니다. 여기에는 Kafka에 연결하고 데이터를 전송하기 위한 KafkaProducer, HTTP GET 요청을 수행해 데이터를 받아오는 requests, .env 파일에 저장된 인증 정보(API 키)를 읽어오는 load_dotenv, 그리고 데이터를 바이트 형태로 전송하기 위한 json 직렬화 모듈이 포함됩니다.

다음으로 os.getenv()를 사용해 API 키를 가져옵니다.

def pull_weather_data():
    cities = ['New York', 'London', 'Johannesburg', 'Nairobi', 'Cairo', 'Doha', 'Tokyo', 'Sydney']
    cities_weather_data = []

    for city in cities:
        url = f"https://api.openweathermap.org/data/2.5/weather?q={city}&appid={API_KEY}"
        response = requests.get(url)
        weather_data = response.json()
        cities_weather_data.append({
            'City' : weather_data['name'],
            'Country' : weather_data['sys']['country'],
            'Temparature' : weather_data['main']['temp'],
            'Humidity' : weather_data['main']['humidity'],
            'Feels_Like' : weather_data['main']['feels_like'],
            'Last_update_time' : weather_data['dt']
        })
    return cities_weather_data

pull_weather_data() 함수는 날씨 정보를 가져와 리스트에 저장합니다.

• 먼저 조회하고 싶은 도시들을 리스트에 정의하고, 수집된 데이터를 담을 빈 리스트 cities_weather_data를 만든다.
• API는 한 번에 하나의 도시만 반환하므로 for 루프를 이용해 도시별로 요청을 보낸다.
• requests.get()으로 HTTP 요청을 수행하고, 응답을 .json()으로 파이썬 딕셔너리로 변환한다.
• 필요한 필드만 추출해 딕셔너리 형태로 만든 뒤 append()로 리스트에 추가한다.
• 최종적으로 완성된 리스트를 반환한다.

producer = KafkaProducer(
    bootstrap_servers = 'localhost:9092',
    value_serializer = lambda v: json.dumps(v).encode('utf-8')
)

KafkaProducer를 사용해 로컬에 실행 중인 Kafka(포트 9092)에 연결한다. Kafka는 데이터를 딕셔너리 형태가 아니라 바이트 형태로 전송하므로, 각 값 v를 json.dumps(v)로 문자열화한 뒤 encode('utf-8')로 바이트로 변환한다.

$ bin/kafka-topics.sh --create --topic open_weather_api_cities_data --bootstrap-server localhost:9092

위 명령으로 Kafka에 토픽을 생성한다.

topic = 'open_weather_api_cities_data'

while True:
    weather_data = pull_weather_data()
    producer.send(topic, weather_data)
    print(f'From openweather: {weather_data}')
    time.sleep(10)

• topic 변수에 데이터를 기록할 토픽 이름을 지정한다.
• while True 무한 루프를 사용해 프로그램이 중단될 때까지 계속 실행한다.
• pull_weather_data()를 호출해 최신 날씨 데이터를 가져오고, producer.send()로 토픽에 전송한다.
• print()를 통해 전송된 데이터를 콘솔에 출력해 시각화 및 디버깅에 활용한다.
• time.sleep(10)으로 10초 대기 후 다시 반복한다.

Producer 코드를 producer.py 파일에 저장하고 실행하면 아래와 같은 출력이 나타납니다.

Consumer 만들기

from kafka import KafkaConsumer
import json

위 코드는 Consumer가 동작하는 데 필요한 의존성을 불러옵니다. 여기에는 Kafka에서 메시지를 읽어오는 KafkaConsumer와 바이트 데이터를 파이썬 딕셔너리로 복원하기 위한 json 역직렬화 모듈이 포함됩니다.

consumer = KafkaConsumer(
    'open_weather_api_cities_data',
    bootstrap_servers = 'localhost:9092',
    auto_offset_reset = 'earliest',
    value_deserializer = lambda m: json.loads(m.decode('utf-8'))
)

• KafkaConsumer()를 이용해 Kafka Consumer 인스턴스를 만든다.
• 읽을 토픽 이름과 Kafka 서버 주소(localhost:9092)를 지정한다.
• auto_offset_reset='earliest' 옵션은 아직 읽지 않은 메시지가 없을 경우 토픽의 가장 처음부터 읽기 시작하도록 한다.
• value_deserializer는 바이트(m)를 UTF‑8 문자열로 디코딩한 뒤 json.loads()로 파이썬 객체로 변환한다.

for message in consumer:
    print(f'Received:{message.value}')

무한 루프를 통해 새로운 메시지가 도착할 때마다 계속 대기하고, 수신된 데이터를 콘솔에 출력한다.

Consumer를 실행했을 때 open_weather_api_cities_data 토픽에서 읽어온 출력 예시는 다음과 같습니다.

---

정리

이 글에서는 Python Producer가 OpenWeather API에서 실시간 날씨 데이터를 수집해 Kafka 토픽에 게시하고, Consumer가 해당 토픽을 지속적으로 읽어와 처리하는 전체 스트리밍 워크플로우를 구현했습니다.

이러한 Producer‑Consumer 구조는 다음과 같은 실제 시스템에서 널리 활용됩니다.

• 금융 거래 모니터링
• 로그 집계
• IoT 텔레메트리
• 사기 탐지
• 실시간 분석
• 이벤트‑드리븐 마이크로서비스

데이터를 신뢰성 있게, 비동기적으로, 대규모로 교환할 수 있기 때문에 다양한 도메인에서 핵심 인프라로 자리 잡고 있습니다.