Einführung in Kafka
Warum wird Kafka verwendet?
Kafka-Architektur
Komponenten:
Produzenten:
Dies sind die Anwendungen oder Dienste, die Daten/Nachrichten an Kafka senden. Produzenten pushen Nachrichten zu bestimmten Themen innerhalb von Kafka.
Themen:
Ein Thema ist eine Kategorie oder ein Feedname, unter dem Datensätze veröffentlicht werden. Die Themen sind partitioniert, um Skalierbarkeit und Parallelität zu ermöglichen.
Partitionen:
Makler:
Verbraucher:
Verbraucher sind Anwendungen oder Dienste, die Nachrichten aus Themen lesen.
Verbraucher abonnieren Themen und beziehen Daten von Kafka-Brokern.
Verbrauchergruppen:
ZooKeeper:
Anwendungsfälle von Kafka
Beispiel mit Python, um zu demonstrieren, wie Kafka in einem Echtzeitszenario verwendet werden kann:
Standortverfolgung für eine Mitfahr-App.
Der Einfachheit halber verwenden wir die Kafka-Python-Bibliothek, um sowohl einen Produzenten (um einen Treiber zu simulieren, der Standortaktualisierungen sendet) als auch einen Verbraucher (um einen Dienst zu simulieren, der diese Standortaktualisierungen verarbeitet) zu erstellen.
1. Kafka einrichten
Stellen Sie sicher, dass Kafka lokal ausgeführt wird, oder nutzen Sie einen Cloud-Anbieter. Sie können Kafka herunterladen und lokal ausführen, indem Sie der Kafka-Schnellstartanleitung folgen.
2. Installieren Sie die Kafka Python-Bibliothek
Sie können die Kafka-Python-Bibliothek mit pip:
installieren
pip install kafka-python
3. Python Kafka Producer (Simulation von Treiberstandortaktualisierungen)
Der Produzent simuliert einen Treiber, der Standortaktualisierungen an ein Kafka-Thema (Treiberstandort) sendet.
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 (Simulierender Ride-Matching-Dienst)
Der Verbraucher liest die Standortaktualisierungen aus dem Treiberstandortthema und verarbeitet sie.
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()
Erklärung:
Produzent (Treiber, der Standortaktualisierungen sendet):
Verbraucher (Mitfahrservice):
Ausführen des Beispiels (ich verwende einen Windows-Rechner):
pip install kafka-python
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)
Führen Sie nun den Produzenten und den Verbraucher in zwei separaten Terminalfenstern mit Python aus.
Führen Sie das Producer-Skript aus, um den Treiber zu simulieren, der Standortaktualisierungen sendet.
Führen Sie das Verbraucherskript aus, um zu sehen, wie der Ride-Matching-Dienst die Standortaktualisierungen in Echtzeit verarbeitet.
Fazit
Apache Kafka bietet eine außergewöhnliche Plattform für die Verwaltung von Echtzeit-Datenströmen. Durch die Kombination von Kafka mit Python können Entwickler leistungsstarke Datenpipelines und Echtzeit-Analyselösungen erstellen.
Ob Fahrzeugverfolgung, IoT-Daten oder Echtzeit-Dashboards, Kafka mit Python ist hoch skalierbar und kann an verschiedene Anwendungsfälle angepasst werden. Beginnen Sie also mit Kafka zu experimentieren und Sie werden von seinem Potenzial in realen Anwendungen begeistert sein!
Das obige ist der detaillierte Inhalt vonEin Leitfaden für Anfänger zu Kafka mit Python: Echtzeit-Datenverarbeitung und -anwendungen. Für weitere Informationen folgen Sie bitte anderen verwandten Artikeln auf der PHP chinesischen Website!
Eigenschaften des Netzwerks
So behalten Sie die Anzahl der Dezimalstellen in C++ bei
So richten Sie die automatische Aktualisierung einer Webseite ein
So lösen Sie ungültige Syntax in Python
So werden Sie auf TikTok ein enger Freund
Freelaunchbar
Webserver
Wie viel entspricht Dimensity 9000 Snapdragon?
![[Web-Frontend] Node.js-Schnellstart](https://img.php.cn/upload/course/000/000/067/662b5d34ba7c0227.png)
