Python을 사용한 Kafka 초보자 가이드: 실시간 데이터 처리 및 애플리케이션

카프카 소개

• Kafka는 Apache에서 개발한 오픈소스 분산 이벤트 스트리밍 플랫폼입니다.
• 원래 LinkedIn에서 만들었으며 높은 처리량, 내결함성 및 실시간 데이터 스트리밍을 처리하도록 설계되었습니다.
• Kafka를 사용하면 시스템에서 레코드(메시지) 스트림을 게시 및 구독하고 처리하고 효율적으로 저장할 수 있습니다.

카프카를 사용하는 이유는 무엇인가요?

• 높은 처리량: Kafka는 초당 수백만 개의 메시지를 처리할 수 있습니다.
• 내결함성: Kafka는 분산되어 있으므로 여러 노드에 걸쳐 데이터를 복제하여 안정성을 보장할 수 있습니다.
• 내구성: Kafka는 데이터를 디스크에 유지하고 메시지를 재생할 수 있어 메시지 전달의 안정성을 보장합니다.
• 실시간 처리: Kafka는 데이터 스트림을 실시간으로 처리할 수 있어 모니터링, 분석 또는 이벤트 기반 시스템과 같은 애플리케이션에 이상적입니다.
• 확장성: Kafka는 대용량 데이터를 처리하기 위해 더 많은 브로커를 추가하여 쉽게 확장할 수 있습니다.
• 분리 시스템: Kafka는 메시징을 위한 중간 계층 역할을 하여 다양한 시스템이 비동기적으로 통신할 수 있도록 해줍니다.

카프카 아키텍처

구성품:

제작자:
Kafka에 데이터/메시지를 보내는 애플리케이션 또는 서비스입니다. 생산자는 Kafka 내의 특정 주제에 메시지를 푸시합니다.

주제:
주제는 레코드가 게시되는 카테고리 또는 피드 이름입니다. 확장성과 병렬 처리가 가능하도록 주제가 분할되어 있습니다.

파티션:

• 각 주제는 하나 이상의 파티션으로 구분됩니다.
• 파티션을 통해 Kafka는 더 많은 메시지와 지원을 처리할 수 있습니다. 병렬 처리.
• 각 파티션에는 고유한 ID가 있으며 파티션의 하위 집합을 저장할 수 있습니다. 주제의 데이터.

브로커:

• Kafka는 각각 데이터를 처리하는 브로커(서버) 클러스터로 실행됩니다. 여러 주제와 파티션에 대해.
• 브로커는 파티션을 저장 및 관리하고 읽기 및 쓰기를 처리합니다. 생산자와 소비자의 요청사항입니다.
• 각 브로커는 고유 ID로 식별됩니다.

소비자:

소비자는 주제의 메시지를 읽는 애플리케이션 또는 서비스입니다.
소비자는 주제를 구독하고 Kafka 브로커에서 데이터를 가져옵니다.

소비자 그룹:

• 소비자는 소비자 그룹으로 구성됩니다.
• 파티션 내의 각 메시지는 그룹 내 한 명의 소비자에게만 전달되므로 여러 소비자 간에 로드 밸런싱이 가능합니다.

동물원 관리인:

• ZooKeeper는 Kafka 브로커를 관리 및 조정하고 브로커, 주제 및 파티션을 추적합니다.
• 파티션의 리더 선택을 관리하고 클러스터 상태를 모니터링하는 데 도움이 됩니다.

Kafka 활용 사례

• 실시간 분석: 기업에서는 금융 거래 분석과 같은 모니터링 시스템을 위해 Kafka를 사용하여 실시간으로 데이터 스트림을 처리하고 분석합니다.
• 로그 집계: Kafka는 처리, 경고 또는 저장을 위해 여러 서비스 또는 애플리케이션의 로그를 통합합니다.
• 데이터 파이프라인: Kafka는 서로 다른 시스템이나 서비스(ETL 파이프라인) 간에 대량의 데이터를 전송하기 위한 백본으로 사용됩니다.
• IoT 애플리케이션: Kafka는 IoT 센서의 데이터 스트림을 처리하여 실시간 분석 및 대응이 가능합니다.
• 마이크로서비스 통신: Kafka는 마이크로서비스 아키텍처를 위한 안정적인 메시징 플랫폼 역할을 하여 비동기식 분리 통신을 가능하게 합니다.
• 실시간 차량 추적: 다음 예는 Kafka를 사용하여 실시간으로 차량을 추적하는 방법을 보여줍니다.

Python을 사용하여 실시간 시나리오에서 Kafka를 사용하는 방법을 보여주는 예:

차량 공유 앱의 위치 추적

간단하게 하기 위해 kafka-python 라이브러리를 사용하여 생산자(위치 업데이트를 보내는 운전자를 시뮬레이션하기 위해)와 소비자(이러한 위치 업데이트를 처리하는 서비스를 시뮬레이션하기 위해)를 모두 생성하겠습니다.

1. 카프카 설정
Kafka가 로컬에서 실행되고 있는지 확인하거나 클라우드 공급자를 사용하세요. Kafka 빠른 시작 가이드에 따라 Kafka를 로컬로 다운로드하고 실행할 수 있습니다.

2. Kafka Python 라이브러리 설치
pip를 사용하여 Kafka Python 라이브러리를 설치할 수 있습니다.

pip install kafka-python

3. Python Kafka Producer(드라이버 위치 업데이트 시뮬레이션)
생산자는 Kafka 주제(driver-location)에 위치 업데이트를 보내는 드라이버를 시뮬레이션합니다.

from kafka import KafkaProducer
import json
import time
import random

# Kafka Producer
producer = KafkaProducer(
    bootstrap_servers=['localhost:9092'],
    value_serializer=lambda v: json.dumps(v).encode('utf-8')  # Serialize data to JSON
)

def send_location_updates(driver_id):
    while True:
        # Simulating random GPS coordinates (latitude, longitude)
        location = {
            "driver_id": driver_id,
            "latitude": round(random.uniform(40.0, 41.0), 6),
            "longitude": round(random.uniform(-74.0, -73.0), 6),
            "timestamp": time.time()
        }
        # Send location data to Kafka
        producer.send('driver-location', location)
        print(f"Sent: {location}")
        time.sleep(5)  # Sleep for 5 seconds to simulate real-time updates

# Start sending updates for driver_id = 101
send_location_updates(driver_id=101)

4. Python Kafka Consumer(승차 매칭 서비스 시뮬레이션)
소비자는 운전자 위치 주제에서 위치 업데이트를 읽고 이를 처리합니다.

from kafka import KafkaConsumer
import json

# Kafka Consumer
consumer = KafkaConsumer(
    'driver-location',
    bootstrap_servers=['localhost:9092'],
    auto_offset_reset='earliest',  # Start from the earliest message
    enable_auto_commit=True,
    group_id='location-group',
    value_deserializer=lambda x: json.loads(x.decode('utf-8'))  # Deserialize data from JSON
)

def process_location_updates():
    print("Waiting for location updates...")
    for message in consumer:
        location = message.value
        driver_id = location['driver_id']
        latitude = location['latitude']
        longitude = location['longitude']
        timestamp = location['timestamp']
        print(f"Received location update for Driver {driver_id}: ({latitude}, {longitude}) at {timestamp}")

# Start consuming location updates
process_location_updates()

설명:

제작자(운전기사가 위치 업데이트를 전송함):

• 생산자는 드라이버 ID, 위도, 경도, 타임스탬프와 같은 필드가 포함된 JSON 개체를 Kafka 주제 드라이버 위치로 보냅니다.
• 제작자는 5초마다 위치 데이터를 전송하여 실시간 GPS 업데이트를 시뮬레이션합니다.

소비자(라이드 매칭 서비스):

• 소비자는 운전자 위치 주제를 구독하고 업데이트를 수신합니다.
• 위치 업데이트가 Kafka에 게시될 때마다 소비자는 이를 처리하고 인쇄하여 이 데이터를 사용하여 운전자와 승객을 일치시키는 서비스를 시뮬레이션합니다.

예제 실행(Windows 컴퓨터에서 실행 중):

1. 사육사 시작

pip install kafka-python

1. 로컬 Kafka 서버를 시작하세요.

from kafka import KafkaProducer
import json
import time
import random

# Kafka Producer
producer = KafkaProducer(
    bootstrap_servers=['localhost:9092'],
    value_serializer=lambda v: json.dumps(v).encode('utf-8')  # Serialize data to JSON
)

def send_location_updates(driver_id):
    while True:
        # Simulating random GPS coordinates (latitude, longitude)
        location = {
            "driver_id": driver_id,
            "latitude": round(random.uniform(40.0, 41.0), 6),
            "longitude": round(random.uniform(-74.0, -73.0), 6),
            "timestamp": time.time()
        }
        # Send location data to Kafka
        producer.send('driver-location', location)
        print(f"Sent: {location}")
        time.sleep(5)  # Sleep for 5 seconds to simulate real-time updates

# Start sending updates for driver_id = 101
send_location_updates(driver_id=101)

이제 Python을 사용하여 2개의 별도 터미널 창에서 생산자와 소비자를 실행합니다.

1. 생산자 스크립트를 실행하여 위치 업데이트를 보내는 운전자를 시뮬레이션합니다.

2. 소비자 스크립트를 실행하여 차량 매칭 서비스가 실시간으로 위치 업데이트를 처리하는 모습을 확인하세요.

결론
Apache Kafka는 실시간 데이터 스트림 관리를 위한 뛰어난 플랫폼을 제공합니다. Kafka와 Python을 결합하면 개발자는 강력한 데이터 파이프라인과 실시간 분석 솔루션을 구축할 수 있습니다.

차량 추적, IoT 데이터, 실시간 대시보드 등 Kafka with Python은 확장성이 뛰어나며 다양한 사용 사례에 맞게 조정할 수 있습니다. 따라서 Kafka를 실험해 보면 실제 애플리케이션에서의 잠재력에 놀라게 될 것입니다!

위 내용은 Python을 사용한 Kafka 초보자 가이드: 실시간 데이터 처리 및 애플리케이션의 상세 내용입니다. 자세한 내용은 PHP 중국어 웹사이트의 기타 관련 기사를 참조하세요!