Einführung
Der Konsolenproducer hat das Ereignis- und Schlüsselmodell veranschaulicht. Eine Anwendung benötigt eine weitere Unterscheidung: Der Aufruf einer Sende-API kann einen Datensatz in eine lokale Clientwarteschlange stellen, bevor der Broker ihn bestätigt. Ein Programm, das an diesem Punkt beendet wird, kann Erfolg melden, ohne seine Datensätze zugestellt zu haben.
Sie verwenden den installierten Python-Client confluent-kafka mit dem vorbereiteten lokalen Apache-Kafka-Broker. Das Paket ist eine Clientbibliothek; ein Confluent-Cloud-Konto ist nicht erforderlich. Sie serialisieren Dictionaries in JSON-/UTF-8-Bytes, senden Bestellschlüssel und untersuchen Zustellungsrückrufe sowie den endgültigen Warteschlangenzustand. Eine nicht erreichbare lokale Adresse ermöglicht einen zeitlich begrenzten Fehlertest.
Dieses unabhängige Lab bereitet ein Topic mit drei Partitionen und zwei Eingabedateien vor. Die späteren Kursaktivitäten setzen Python-Funktionen, Schleifen, Dictionaries, JSON und das Ausführen von Skripten voraus. Der Python-Lernpfad kann Ihnen bei diesen Voraussetzungen helfen. Kafka-Installation und Dienstkonfiguration sind weiterhin vorbereitet. Verwenden Sie das Terminal.
Wandeln Sie ein Dictionary in einen zu übertragenden Schlüssel und Wert um
In diesem Schritt untersuchen Sie die Umwandlung, bevor Sie Kafka kontaktieren. Beginnen Sie mit der Eingabe aus sechs Ereignissen:
cd /home/labex/project
cat /home/labex/project/data/events.jsonl
Jede Zeile ist ein vollständiges JSON-Objekt. json.loads wandelt diesen Text in ein Python-Dictionary um. json.dumps wandelt das Dictionary in JSON-Text um, und encode("utf-8") wandelt Text in Bytes um, die für die Client-API geeignet sind. Der Schlüssel wird separat aus order_id kodiert; er ist nicht die event_id.
Speichern Sie dieses kleine Vorschauprogramm und führen Sie es aus. Das in Anführungszeichen begrenzte Here-Dokument erhält den Python-Quelltext wörtlich:
cat > /home/labex/project/preview_event.py <<'PYTHON'
import json
from pathlib import Path
event = json.loads(Path("/home/labex/project/data/events.jsonl").read_text().splitlines()[0])
value_text = json.dumps(event, ensure_ascii=False)
value_bytes = value_text.encode("utf-8")
key_bytes = event["order_id"].encode("utf-8")
print("Event dict:", event)
print("JSON text:", value_text)
print("UTF-8 value:", value_bytes)
print("UTF-8 key:", key_bytes)
PYTHON
bigdata-python /home/labex/project/preview_event.py > /home/labex/project/output/encoding-preview.txt
cat /home/labex/project/output/encoding-preview.txt
Suchen Sie das Dictionary, den JSON-Text und die Bytedarstellungen. Eine Python-Bytes-Darstellung beginnt mit b; diese Darstellungsnotation ist nicht Teil des übertragenen Werts. Der erste Schlüssel sollte order-4001 kodieren. JSON erhält die benannten Ereignisfelder, während UTF-8 die Zeichenkodierung bereitstellt.
Es wurde noch kein Ereignis gesendet. Im nächsten Schritt übergeben Sie diese Bytes an einen Producer und prüfen, wann der Broker sie bestätigt.
Veröffentlichen und bestätigen Sie jede tatsächliche Zustellung
In diesem Schritt führen Sie einen vollständigen Producer mit bereitgestellter Argumentverarbeitung und einem Zustellungsrückruf aus. Producer konfiguriert den Client; bootstrap.servers gibt die anfängliche Adresse an. acks=all fordert eine Bestätigung von den aktuellen synchronen Replikaten an. Diese Umgebung mit einem Broker hat ein Replikat. Daher demonstriert diese Einstellung keine Ausfallsicherheit über mehrere Broker hinweg. Die bereitgestellte Weiterleitungseinstellung murmur2_random hält die schlüsselbasierte Zuordnung mit den früheren Java-Konsolenbeispielen kompatibel, solange die Partitionsanzahl unverändert bleibt.
Beachten Sie die zentrale Unterscheidung im Programm: produce stellt einen Datensatz lokal in die Warteschlange. poll verarbeitet Rückrufe, und flush wartet auf ausstehende Arbeit und verarbeitet dabei ebenfalls Rückrufe.
Ein Callback ist eine Funktion, die Sie der Clientbibliothek übergeben, damit diese sie aufruft, sobald ein Zustellungsergebnis verfügbar ist. Mit callback=delivery übergeben Sie die Funktion selbst; Sie rufen sie nicht sofort auf. poll(0) verarbeitet verfügbare Rückrufe ohne zu warten, während flush zusätzlich auf ausstehende Sendevorgänge wartet. Der Callback zeichnet entweder ein vom Broker bestätigtes Topic samt Partition und Offset oder einen Fehler auf. Nach flush prüft das Programm verbleibende Datensätze, Fehler und die bestätigte Anzahl, bevor es Erfolg meldet. Ein Fehlerrückruf wird gesammelt und führt zu einem Prozessergebnis ungleich null; er wird nicht nur ausgegeben und anschließend ignoriert.
Der bereitgestellte Abschnitt argparse ordnet Befehlszeilenoptionen den Attributen von args zu: Beispielsweise wird --input zu args.input und --delivery-timeout-ms zu args.delivery_timeout_ms. Path(args.input).open(...) öffnet die ausgewählte lokale Datei. Die Schleife liest ein JSON-Ereignis je nicht leerer Zeile. Sie müssen keinen Argumentparser entwerfen, um diesen Client zu verwenden.
Speichern Sie das vollständige Programm. Die Parser- und Callback-Struktur ist vorgegeben, während die Serialisierung und der produce-Aufruf die gerade untersuchten Bytes verwenden:
cat > /home/labex/project/producer.py <<'PYTHON'
import argparse,json,sys
from pathlib import Path
from confluent_kafka import Producer
parser = argparse.ArgumentParser(description="Publish a finite JSONL event file and confirm deliveries.")
parser.add_argument("--input", required=True)
parser.add_argument("--topic", required=True)
parser.add_argument("--bootstrap-server", default="localhost:9092")
parser.add_argument("--delivery-timeout-ms", type=int, default=15000)
args = parser.parse_args()
if args.delivery_timeout_ms <= 0:
parser.error("The delivery timeout must be positive.")
producer = Producer({
"bootstrap.servers": args.bootstrap_server,
"acks": "all",
"message.timeout.ms": args.delivery_timeout_ms,
"partitioner": "murmur2_random",
})
delivered = []
failures = []
def delivery(error, message):
event_id = json.loads(message.value().decode("utf-8"))["event_id"]
if error is not None:
failures.append(str(error))
print(f"DELIVERY_FAILED event_id={event_id} error={error}", file=sys.stderr)
else:
delivered.append(event_id)
print(f"DELIVERED event_id={event_id} topic={message.topic()} partition={message.partition()} offset={message.offset()}")
sent = 0
with Path(args.input).open(encoding="utf-8") as source:
for line in source:
if not line.strip():
continue
event = json.loads(line)
key = event["order_id"].encode("utf-8")
value = json.dumps(event, ensure_ascii=False).encode("utf-8")
producer.produce(args.topic, key=key, value=value, callback=delivery)
sent += 1
producer.poll(0)
remaining = producer.flush(args.delivery_timeout_ms / 1000 + 5)
if remaining or failures or len(delivered) != sent:
print(f"Delivery incomplete: queued={remaining}, failures={len(failures)}, confirmed={len(delivered)}, requested={sent}", file=sys.stderr)
raise SystemExit(1)
print("Delivered records:", len(delivered))
PYTHON
Führen Sie den Producer aus und speichern Sie beide Ausgabeströme. > leitet die Standardausgabe in das Protokoll um; 2>&1 sendet die Standardfehlerausgabe an dasselbe Ziel, sodass auch Callback-Fehler und Verbindungsdiagnosen erhalten bleiben.
bigdata-python /home/labex/project/producer.py --input /home/labex/project/data/events.jsonl --topic orders-python > /home/labex/project/output/producer-deliveries.log 2>&1
cat /home/labex/project/output/producer-deliveries.log
Erwarten Sie sechs DELIVERED-Callback-Zeilen und Delivered records: 6. Die tatsächliche Partition und der Offset jedes Rückrufs können abweichen. Eine erfolgreiche Rückkehr von produce allein würde diese Bestätigungen nicht belegen.
Untersuchen Sie den Broker unabhängig mit dem bereits erlernten Konsolenconsumer:
kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic orders-python --from-beginning --max-messages 6 --timeout-ms 30000 --formatter-property print.key=true --formatter-property print.partition=true --formatter-property print.offset=true > /home/labex/project/output/python-first-observed.txt
cat /home/labex/project/output/python-first-observed.txt
Suchen Sie py1 bis py6 mit Schlüsseln, die ihren order_id-Werten entsprechen. Der unabhängige Lesevorgang verbindet den Zustellungsbericht der Anwendung mit tatsächlichen Brokerdatensätzen. Eine feste Partitionsnummer oder globale Anzeigereihenfolge ist nicht erforderlich.
Ändern Sie ein Feld in einem neuen Ereignis und verwenden Sie den Producer erneut
In diesem Schritt verwenden Sie dieselbe Anwendung für ein neues Ereignis. Die vorbereitete Datei next-event hat eine neue event_id, py7, und wurde daher noch nicht veröffentlicht. Untersuchen Sie ihr Feld channel:
cat /home/labex/project/data/next-event.jsonl
Ändern Sie nur den channel-Wert dieses unveröffentlichten Ereignisses von online zu store in einer separaten Arbeitsdatei. Das Ereignis bezieht sich weiterhin auf order-4001 und folgt mit sequence 3 auf deren frühere Ereignisse:
cat > /home/labex/project/data/changed-event.jsonl <<'EVENT'
{"event_id":"py7","order_id":"order-4001","sequence":3,"event_type":"shipped","channel":"store"}
EVENT
bigdata-python /home/labex/project/producer.py --input /home/labex/project/data/changed-event.jsonl --topic orders-python > /home/labex/project/output/changed-delivery.log 2>&1
cat /home/labex/project/output/changed-delivery.log
kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic orders-python --from-beginning --max-messages 7 --timeout-ms 30000 --formatter-property print.key=true --formatter-property print.partition=true --formatter-property print.offset=true > /home/labex/project/output/python-final-observed.txt
cat /home/labex/project/output/python-final-observed.txt
Erwarten Sie einen erfolgreichen neuen Rückruf und sieben gespeicherte Ereignisse im unabhängigen Lesevorgang. py7 sollte channel store haben, während py1 bis py6 ihren ursprünglichen Inhalt behalten. Zusammengehörige Ereignisse von order-4001 folgen ihrer Abfolge innerhalb derselben zugeordneten Partition. Die Anzeigereihenfolge über Partitionen hinweg kann abweichen.
Die Producer-Implementierung wurde nicht geändert. Ein übergebener Eingabepfad ermöglichte ihr, die neuen Daten zu serialisieren und zuzustellen, statt eine fest eingebaute Übungsantwort zu verwenden.
Beobachten Sie einen zeitlich begrenzten Fehler an einer nicht erreichbaren Adresse
In diesem Schritt prüfen Sie, dass das Einreihen in die Warteschlange nicht mit der Zustellung verwechselt wird. Derselbe Client kann eine lokale Warteschlange erstellen, auch wenn unter localhost:1 kein Kafka-Broker vorhanden ist. Die bereitgestellte Option für die Zustellungsfrist begrenzt den Nachrichtenabschluss auf 3000 Millisekunden, und flush fügt einen kleinen begrenzten Zeitpuffer hinzu. Der tatsächliche Broker bleibt unter localhost:9092 erreichbar.
Senden Sie die Datei mit einem Ereignis an diese absichtlich falsche Adresse. Der nächste Befehl soll fehlschlagen.
bigdata-python /home/labex/project/producer.py --input /home/labex/project/data/changed-event.jsonl --topic orders-python --bootstrap-server localhost:1 --delivery-timeout-ms 3000 > /home/labex/project/output/unreachable-producer.log 2>&1
Erfassen Sie den Status sofort, bevor ihn ein anderer Befehl ersetzt. $? ist der Exitstatus des vorherigen Befehls: null bei Erfolg, ungleich null bei einem Fehler. printf speichert ihn als Textzeile:
printf '%s\n' "$?" > /home/labex/project/output/unreachable-exit-status.txt
cat /home/labex/project/output/unreachable-exit-status.txt
cat /home/labex/project/output/unreachable-producer.log
Erwarten Sie einen Status ungleich null, Verbindungsdiagnosen, einen DELIVERY_FAILED-Rückruf und Delivery incomplete statt einer erfolgreichen abschließenden Anzahl. Das Programm hat Callback-Fehler geprüft, auch wenn die Warteschlange nach fehlgeschlagenen Nachrichten vollständig geleert war.
Die fehlgeschlagene Adresse sollte kein zweites py7 zum tatsächlichen Topic hinzugefügt haben. Alle sieben früher erfolgreich zugestellten Ereignisse bleiben verfügbar. Einen Endpunkt zu korrigieren und ein Geschäftsereignis erneut zu senden, ist eine andere Entscheidung als anzunehmen, dass ein eingereihter Datensatz bestätigt wurde. Der absichtliche Fehler dieses Labs hat nichts an den tatsächlichen Broker gesendet. Als Nächstes dekodieren Sie empfangenes JSON, speichern es und schreiben den Lesefortschritt in einem Python-Consumer fest.
Zusammenfassung
Sie haben Dictionaries und Bestellschlüssel als UTF-8-Bytes serialisiert, über einen tatsächlichen Python-Client veröffentlicht und erfolgreiche Rückrufe mit tatsächlichen Brokerpositionen abgeglichen. Ein geändertes neues Ereignis verwendete dasselbe Programm erneut, während eine nicht erreichbare Adresse zu einem zeitlich begrenzten Fehler mit Status ungleich null statt zu einem falschen Erfolg führte. Als Nächstes macht ein Python-Consumer das Dekodieren, die Dateiausgabe und das Schreiben vor dem Festschreiben explizit.
