Einführung
Ein Consumer, der ein Ereignis anzeigt, hat damit noch nicht unbedingt eine nützliche Übergabe gespeichert. Diese Anwendung dekodiert jeden Wert, erhält seinen Schlüssel und seine Kafka-Position in einer lokalen JSONL-Datei und schreibt den Fortschritt erst fest, nachdem der zugehörige Schreibvorgang erfolgreich war und der Schreibpuffer geleert wurde.
Sie übertragen das bekannte Gruppen- und Offsetmodell auf Python: subscribe wählt das Topic aus, poll fordert Datensätze an und ein expliziter synchroner Commit speichert den nächsten Offset für die verarbeitete Nachricht. Automatisches Festschreiben und automatische Offsetablage sind deaktiviert, damit sie den Fortschritt nicht unabhängig von Ihrem Code voranbringen. Dateischreiben und Kafka-Commits bleiben getrennte Vorgänge; dies ist keine atomare systemübergreifende Transaktion.
Dieses unabhängige Lab bereitet die sechs anfänglichen Bestellereignisse aus dem Producer-Lab in einem neuen Topic vor. Ein VM-Zustand aus früheren Aktivitäten ist nicht erforderlich. Sie benötigen Python-Funktionen, Schleifen, Dictionaries, JSON und Kenntnisse zum Ausführen von Skripten. Die Parser- und Friststruktur ist vorgegeben. Verwenden Sie das Terminal und die installierte bigdata-python-Umgebung.
Abonnieren, pollen und dekodieren Sie einen tatsächlichen Datensatz
In diesem Schritt untersuchen Sie einen empfangenen Datensatz, ohne den Fortschritt der Vorschau festzuschreiben. Lesen Sie zuerst die vorbereiteten Ereigniswerte:
cd /home/labex/project
cat /home/labex/project/data/events.jsonl
Ein Consumer wird mit einer Gruppenidentität und einer anfänglichen earliest-Richtlinie konfiguriert. subscribe registriert das gewünschte Topic. poll gibt entweder eine Nachricht, eine Nachricht mit Fehler oder None zurück, solange kein Datensatz verfügbar ist, auch während der Zuweisung. Die Schleife fragt innerhalb einer Gesamtfrist weiter ab.
Die vorgegebene Kontrollflussstruktur begrenzt diesen Lesevorgang zeitlich. time.monotonic() misst verstrichene Zeit unabhängig vom Kalenderdatum. continue beginnt die nächste Abfrage, wenn keine Nachricht bereitsteht; break verlässt die Schleife, nachdem ein Datensatz angezeigt wurde. Das else der while-Schleife wird nur ausgeführt, wenn die Schleife ohne dieses break endet. Hier bedeutet das, dass vor Ablauf der Frist kein Datensatz eingetroffen ist.
finally schließt den Consumer, unabhängig davon, ob der Lesevorgang erfolgreich ist oder einen Fehler auslöst. Verstehen Sie dieses Gerüst als Warten und Aufräumen rund um die Kafka-Aufrufe. Sie müssen kein System für Wiederholungsversuche entwerfen.
Diese Vorschau verwendet eine separate Gruppe file-preview, bei der beide automatischen Einstellungen deaktiviert sind. Der dekodierte Wert ist ein Python-Dictionary; die Hülle erhält außerdem Topic, Partition, Offset und einen dekodierten Schlüssel. Es erfolgt kein commit-Aufruf:
cat > /home/labex/project/preview_consumer.py <<'PYTHON'
import json,time
from confluent_kafka import Consumer
consumer = Consumer({"bootstrap.servers":"localhost:9092", "group.id":"file-preview", "auto.offset.reset":"earliest", "enable.auto.commit":False, "enable.auto.offset.store":False, "allow.auto.create.topics":False})
deadline = time.monotonic() + 20
try:
consumer.subscribe(["orders-file"])
while time.monotonic() < deadline:
message = consumer.poll(1)
if message is None:
continue
if message.error() is not None:
raise RuntimeError(str(message.error()))
value = json.loads(message.value().decode("utf-8"))
record = {"topic":message.topic(), "partition":message.partition(), "offset":message.offset(), "key":message.key().decode("utf-8"), "value":value}
print(json.dumps(record, ensure_ascii=False))
break
else:
raise SystemExit("No preview record arrived before the deadline.")
finally:
consumer.close()
PYTHON
bigdata-python /home/labex/project/preview_consumer.py > /home/labex/project/output/one-event.json
cat /home/labex/project/output/one-event.json
Erwarten Sie eine JSON-Hülle, deren value eines der Ereignisse py1 bis py6 enthält und deren key der order_id dieses Ereignisses entspricht. Bei drei Partitionen steht das zuerst beobachtete Ereignis nicht fest. Partition und Offset identifizieren seinen tatsächlichen Brokerdatensatz.
Das Schließen dieser Vorschau schreibt keinen Fortschritt fest, da automatische Commits deaktiviert sind und das Skript keinen expliziten Commit ausführt. Die nächste Anwendung verwendet für die Dateiübergabe eine eigene Gruppe.
Schreiben Sie jede Hülle und leeren Sie den Puffer vor dem Commit ihrer Nachricht
In diesem Schritt erstellen Sie den begrenzten Datei-Consumer. Lesen Sie seine Verarbeitungsreihenfolge sorgfältig: JSON-Objekt und Schlüssel dekodieren und prüfen, eine JSONL-Hülle schreiben, den Dateipuffer leeren, diese Nachricht synchron festschreiben und sie anschließend als abgeschlossen zählen. Wenn Schreiben, Dekodieren oder Commit fehlschlägt, beendet sich die Anwendung, statt den Fehler zu überspringen und später einen erfolgreichen Stapel vorzutäuschen.
Die with-Anweisung schließt die lokale Datei beim Verlassen ihres Blocks. Path bezeichnet den lokalen Dateinamen. Die Ausgabe wird im Anhängemodus geöffnet, damit ein Neustart frühere Zustellungsnachweise nicht stillschweigend abschneidet. Der saubere erste Lauf beginnt ohne Datei. max-messages ist das Ziel für diesen Aufruf; die Gesamtfrist sorgt dafür, dass ein unvollständiger Lauf mit Status ungleich null endet. JSONL speichert ein vollständiges JSON-Objekt je Zeile ohne umgebendes Array. flush() überträgt die gepufferten Dateidaten von Python vor der Commit-Anforderung an das Betriebssystem. Es garantiert nicht das Überstehen eines Strom- oder Datenträgerausfalls und macht aus der Datei und dem Kafka-Commit keine atomare Transaktion. Diese Übung beobachtet eine Prozessunterbrechung und den gespeicherten Fortschritt in der laufenden VM.
commit(message=..., asynchronous=False) schreibt den nächsten Offset der verarbeiteten Nachricht für ihre Partition fest. Die zurückgegebenen Fehler je Partition werden geprüft. enable.auto.offset.store=false verhindert eine unabhängige automatische Fortschrittsablage, während das explizite Argument message genau die verarbeitete Position identifiziert, die festgeschrieben wird.
Speichern Sie die vollständige vorgegebene Parser- und Schleifenstruktur mit dieser Verarbeitungsfolge:
cat > /home/labex/project/consumer.py <<'PYTHON'
import argparse,json,sys,time
from pathlib import Path
from confluent_kafka import Consumer
parser = argparse.ArgumentParser(description="Save a finite JSON event delivery and commit only written records.")
parser.add_argument("--topic", required=True)
parser.add_argument("--group", required=True)
parser.add_argument("--output", required=True)
parser.add_argument("--max-messages", type=int, default=6)
parser.add_argument("--timeout-seconds", type=float, default=30)
parser.add_argument("--bootstrap-server", default="localhost:9092")
args = parser.parse_args()
if args.max_messages <= 0 or args.timeout_seconds <= 0:
parser.error("Message count and timeout must be positive.")
consumer = Consumer({
"bootstrap.servers": args.bootstrap_server,
"group.id": args.group,
"auto.offset.reset": "earliest",
"enable.auto.commit": False,
"enable.auto.offset.store": False,
"allow.auto.create.topics": False,
})
received = 0
deadline = time.monotonic() + args.timeout_seconds
try:
consumer.subscribe([args.topic])
with Path(args.output).open("a", encoding="utf-8") as sink:
while received < args.max_messages and time.monotonic() < deadline:
message = consumer.poll(1)
if message is None:
continue
position = f"topic={message.topic()} partition={message.partition()} offset={message.offset()}"
if message.error() is not None:
raise RuntimeError(f"Kafka error {position}: {message.error()}")
try:
key = message.key().decode("utf-8") if message.key() is not None else None
value = json.loads(message.value().decode("utf-8"))
if not isinstance(value, dict) or not isinstance(value.get("event_id"), str) or not isinstance(value.get("order_id"), str):
raise ValueError("Expected an object with string event_id and order_id.")
if key != value["order_id"]:
raise ValueError("Record key does not match order_id.")
except (UnicodeDecodeError, json.JSONDecodeError, ValueError) as error:
print(f"Decode failure {position}: {error}", file=sys.stderr)
raise SystemExit(1)
record = {"topic": message.topic(), "partition": message.partition(), "offset": message.offset(), "key": key, "value": value}
sink.write(json.dumps(record, ensure_ascii=False) + "\n")
sink.flush()
committed = consumer.commit(message=message, asynchronous=False)
if not committed or any(item.error is not None for item in committed):
raise RuntimeError(f"Commit failed {position}: {committed}")
received += 1
print(f"SAVED_AND_COMMITTED event_id={value['event_id']} {position} next_offset={message.offset()+1}")
if received < args.max_messages:
print(f"Only received {received} of {args.max_messages} requested records before the deadline.", file=sys.stderr)
raise SystemExit(2)
print("Saved and committed records:", received)
finally:
consumer.close()
PYTHON
bigdata-python /home/labex/project/consumer.py --topic orders-file --group file-writer --output /home/labex/project/output/received-events.jsonl --max-messages 6 --timeout-seconds 30 > /home/labex/project/output/consumer.log 2>&1
cat /home/labex/project/output/consumer.log
cat /home/labex/project/output/received-events.jsonl
Erwarten Sie Saved and committed records: 6 und sechs JSONL-Hüllen. Jede Hülle sollte den vollständigen Wert, den passenden Schlüssel und die tatsächliche Datensatzposition erhalten. Ein Callback oder Konsolenprotokoll ersetzt die gespeicherte Datei nicht. Der nächste Schritt vergleicht Dateinachweise mit dem tatsächlich festgeschriebenen Fortschritt.
Verbinden Sie die Datei mit den nächsten festgeschriebenen Offsets je Partition
In diesem Schritt untersuchen Sie den Fortschritt des Koordinators mit dem bereits bekannten Befehlszeilenwerkzeug:
kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group file-writer --describe > /home/labex/project/output/file-writer-position.txt
cat /home/labex/project/output/file-writer-position.txt
Für jede Partition mit zugestellten Datensätzen sollte CURRENT-OFFSET den LOG-END-OFFSET dieser Partition bei Rückstand null erreichen. Genaue Partitionszuweisungen und die Verteilung werden in dieser Lektion nicht festgelegt. Eine Partition ohne Datensätze hat möglicherweise noch keine festgeschriebene Position. Dort gibt es keine verarbeitete Nachricht, die festgeschrieben werden könnte.
Vergleichen Sie den Offset einer gespeicherten Hülle mit dem festgeschriebenen nächsten Offset ihrer Partition. Wenn der letzte verarbeitete Datensatz einer Partition bei Offset n liegt, ist die nächste festgeschriebene Position n+1. Diese nächste Position ist keine globale Anzahl über alle Partitionen hinweg.
Der Consumer wurde geschlossen, daher ist eine Meldung über fehlende aktive Mitglieder normal. Die gespeicherte Datei und der gespeicherte Gruppenfortschritt bleiben erhalten. Sie haben getrennte Nachweise der Verarbeitungsausgabe und des Koordinatorzustands, ohne zu behaupten, dass sie in einer Transaktion geschrieben wurden.
Starten Sie dieselbe Gruppe neu und beobachten Sie das Ausbleiben neuer Zustellungen
In diesem Schritt testen Sie die gespeicherte Position, statt anzunehmen, dass earliest immer zurückspult. Die sechs Datensätze werden weiterhin aufbewahrt, doch file-writer hat gültige festgeschriebene Endpositionen. Überlegen Sie, was ein weiterer Start derselben Gruppe ergibt: Sie sollte keine neuen Datensätze finden, obwohl auto.offset.reset auf earliest steht.
Verwenden Sie einen separaten Ausgabepfad, um die ursprüngliche Datei mit sechs Datensätzen zu erhalten. Fordern Sie einen Datensatz mit einer kurzen Gesamtfrist an. Da die angeforderte Anzahl nicht erreicht werden kann, meldet diese Anwendung bewusst einen Status ungleich null für einen unvollständigen Stapel. Der nächste Befehl soll fehlschlagen.
bigdata-python /home/labex/project/consumer.py --topic orders-file --group file-writer --output /home/labex/project/output/no-new-events.jsonl --max-messages 1 --timeout-seconds 5 > /home/labex/project/output/same-group-restart.log 2>&1
Erfassen Sie seinen Status sofort:
printf '%s\n' "$?" > /home/labex/project/output/same-group-exit-status.txt
cat /home/labex/project/output/same-group-exit-status.txt
cat /home/labex/project/output/same-group-restart.log
Erwarten Sie einen Status ungleich null und Only received 0 of 1 requested records before the deadline. Die neue Datei sollte leer sein; die ursprüngliche received-events.jsonl enthält weiterhin sechs Hüllen. Die gültigen festgeschriebenen Offsets derselben Gruppe wurden nicht durch ihre Startvorgabe earliest ersetzt.
Um gespeicherte Verläufe absichtlich erneut zu lesen, verwenden Sie den bereits gezeigten Ablauf mit einer anderen Gruppe oder dem Zurücksetzen einer inaktiven Gruppe. Wenn ein Prozess nach dem Schreiben, aber vor dem Commit stoppt, ist die Situation anders: Dieselbe Gruppe kann den Datensatz erneut empfangen. Das nächste Lab macht dieses Fehlerfenster sichtbar.
Zusammenfassung
Sie haben mit Python abonniert und gepollt, JSON-/UTF-8-Datensätze dekodiert und Schlüssel sowie Kafka-Positionen in JSONL erhalten. Erfolgreiches Leeren der Schreibpuffer ging expliziten synchronen Commits voraus, und der tatsächliche Koordinatorzustand stimmte mit der abgeschlossenen Datei überein. Ein Neustart derselben Gruppe fand keine neuen Datensätze, ohne den gespeicherten Topicverlauf zu löschen. Als Nächstes zeigen ein Dekodierungsfehler und eine Unterbrechung zwischen Schreiben und Commit, warum Datensätze erneut zugestellt werden können.
