Einführung
In einer kleinen Galerie können Besucher für ihren Lieblingsstand stimmen. Ein Sender kann ein Ereignis erneut übertragen. Ein neuer Kafka-Datensatz bedeutet deshalb nicht immer eine zusätzliche Stimme. Du untersuchst zunächst ein normales Beispiel mit zwei Stimmen, dann eine echte erneute Übertragung an einer anderen gespeicherten Position und rekonstruierst schließlich die Stimmenzahlen der Galerie.
Vorausgesetzt werden Terminalpfade, Python-Funktionen/Schleifen/Dictionaries, JSON sowie die Grundlagen zu Kafka-Schlüsseln, Partitionen, Offsets, Consumer-Gruppen und dem Python-Client aus Apache Kafka for Beginners. Dieses Lab ergänzt die Identität eines Geschäftsereignisses. Ein vollständiger nativer Leser ist vorbereitet, damit du dich auf die Zählregel konzentrieren kannst; im nächsten angeleiteten Lab baust du diesen Leser selbst. Verwende Terminal und die installierte Umgebung bigdata-python. Die Schätzung des Autors von 40–50 Minuten umfasst Lesen und gewöhnliche Fehler, ist aber keine gemessene Anfänger-Erprobung.
Es geht nur um einen endlichen, bereits zugestellten Batch. Weder ein System zur Prüfung der Wahlberechtigung noch ein kontinuierlicher Stream, eine garantierte Wiederherstellung nach Abstürzen oder Ende-zu-Ende-Exactly-once-Verarbeitung sind Bestandteil. Jedes Lab startet unabhängig.
Beobachte eine erneute Übertragung an einer anderen gespeicherten Position
In diesem Schritt vergleichst du zwei normale Stimmen mit einem erneut übertragenen Ereignis. Beginne in Terminal. Im Arbeitsverzeichnis liegen vorbereitete Werkzeuge und Kataloge. Bearbeite nur reducer.py, replay.py und optionale Python-Helfer unter helpers/; abgeleitete Dateien gehören unter output/. Bewahre die bereitgestellten Werkzeuge/Daten, die eingefrorenen Quell-Topics und die Gruppen der pausierten Anzeigeanwendungen. Im separaten Demo-Topic ist genau eine Aktion erlaubt: die folgende erneute Übertragung.
cd /home/labex/project
-B verhindert, dass diese Python-Befehle Bytecode-Cachedateien erzeugen. Das nur lesende Werkzeug inspect zeigt die tatsächlichen Datensatzhüllen des Brokers: Topic, Partition, Offset, Schlüssel und dekodierten Wert. Die abschließende Anzahl zählt gespeicherte Anwendungsdatensätze. Untersuche die kleine Demo:
bigdata-python -B vote_tool.py inspect demo
Erwartet werden zwei Datensätze. Ihre Werte haben dieses Schema: event_id, booth und choice sind nichtleere Zeichenfolgen; choice ist genau favourite; booth muss im ausgewählten Katalog stehen. Zusätzliche Felder gehören nicht zum Stimmenschema dieses Kurses. Der Schlüssel ist der Standname und dient dem Routing. event_id identifiziert eine Stimme auf fachlicher Ebene. Hier stimmt demo-1 für oak und demo-2 für fern: zwei Datensätze und zwei logische Stimmen.
Sage vor dem Senden beide Ergebnisse voraus: Wie viele physische Datensätze werden gespeichert, und wie lauten die Standzahlen, wenn demo-1 weiterhin eine einzige fachliche Stimme ist? Prüfe dann diese Vorhersage.
Der Wiederholungsbefehl sendet denselben vollständigen Wert von demo-1 an den echten Broker und wartet auf die Zustellbestätigung. Vor dem Senden prüft er die vorhandene Demo. Eine weitere Ausführung nach der erwarteten Wiederholung hängt deshalb keinen vierten Datensatz an:
bigdata-python -B vote_tool.py retry-demo
bigdata-python -B vote_tool.py inspect demo
Jetzt gibt es drei Datensätze. demo-1 steht an zwei unterschiedlichen Offsets mit identischem Wertinhalt. Diese Positionen identifizieren zwei gespeicherte Datensätze; die wiederholte Ereignis-ID identifiziert weiterhin nur eine Stimme. Die richtigen Standzahlen bleiben oak=1, fern=1, reed=0. Einen vorhandenen Datensatz erneut zu lesen und eine weitere Übertragung zu speichern sind unterschiedliche Situationen. Beide benötigen beim Zählen eine Regel für die fachliche Identität.
Der eingefrorene Kurs-Batch nutzt eine explizite Platzierung in Partitionen. Gleiche Standschlüssel können deshalb in verschiedenen Partitionen auftreten. Lies beim Vergleich von Datensätzen die tatsächlichen Metadaten.
Innerhalb des ausgewählten Kurs-Batches gilt eine Ereignis-ID global über alle Partitionen, nicht nur innerhalb einer Partition oder eines Stands. Zwei verschiedene IDs zählen als zwei Stimmen, selbst wenn alle anderen Felder gleich sind. Die vorgegebenen fachlichen Regeln akzeptieren diese Ereignisse; sie legen nicht fest, wie viele Stimmen eine reale Person abgeben darf.
Meldet ein Werkzeug einen unbekannten Batch, liste die vorbereiteten Namen auf und verwende für diese Beobachtung demo:
bigdata-python -B vote_tool.py list
Ein Tippfehler hat den Broker nicht zurückgesetzt. Korrigiere das Argument und wiederhole die nur lesende Prüfung. Sende keine beliebigen Ereignisse erneut, um eine angezeigte Stimmenzahl zu reparieren.
Bewahre logische Identitäten und lehne widersprüchliche Inhalte ab
In diesem Schritt implementierst du die fachliche Regel innerhalb der vorbereiteten Python-Funktion. Bleibe in Terminal unter /home/labex/project. Lies die aktuelle Datei:
cat reducer.py
Sie initialisiert die Katalogzähler und erhöht sie dann für jeden Eingabewert. Sage voraus, was sie mit den drei Demo-Datensätzen tun würde: Der zweite gespeicherte demo-1 würde eine weitere Stimme hinzufügen. Die richtige Regel muss diese ID einmal zählen und unterschiedliche IDs dabei getrennt halten.
Der vorbereitete Regelprüfer ruft eine exakte Kopie deiner Funktion mit wegwerfbaren Beispielen auf, darunter umgeordnete Werte und andere Standnamen. Er verändert das aktive Kafka-Log nicht. Führe die aktuelle naive Funktion aus; dieser Befehl soll mit einer Meldung zur Regel fehlschlagen:
bigdata-python -B check_rules.py
Die Shell speichert den Exitstatus des unmittelbar vorherigen Befehls in $?: Null bedeutet Erfolg, ein anderer Wert einen Fehler. Zeige ihn sofort an, bevor ein anderer Befehl ihn ersetzt:
printf '%s\n' "$?"
summarize(values, booths) muss bei gültiger Eingabe ein Dictionary mit votes und counts zurückgeben und bei ungültiger Eingabe eine gewöhnliche Exception auslösen. Das Modul muss aufrufbar bleiben; beende darin nicht den Python-Interpreter. Der Prüfer muss tatsächliche Rückgabewerte erhalten, nicht bloß einen Exitstatus des Interpreters.
Bearbeite die vorhandene Funktion in nano
Öffne die bearbeitbare Datei im Texteditor des Terminals:
nano reducer.py
Die obere Leiste zeigt den Dateinamen. Bewege dich mit den Pfeiltasten und korrigiere Zeichen mit Rücktaste/Entf. Python gruppiert Anweisungen durch Einrückung: Verwende vier Leerzeichen innerhalb der Funktion, acht innerhalb ihrer Schleife und weitere vier für jede verschachtelte Bedingung. Erhalte die vorhandene Zeile def summarize(values, booths):. Entferne den alten Kommentar über das absichtliche Zählen von Datensätzen, wenn du diese Logik ersetzt.
Die folgenden kurzen Python-Blöcke sind zusammenhängende Änderungen innerhalb dieser einen Funktion, keine an der Shell auszuführenden Befehle und keine getrennten Programme. Vervollständige alle Änderungen, bevor du den Prüfer ausführst. Bewahre die gültigen Teile des Starters, statt die ganze Datei zu ersetzen.
Erhalte zuerst die vorhandene Initialisierung der Zähler mit null und füge seen unmittelbar darunter vor der Schleife hinzu. Ein Dictionary mit event_id als Schlüssel bewahrt den vollständigen Wert jeder akzeptierten fachlichen Stimme:
counts = {booth: 0 for booth in booths}
seen = {}
Erhalte als Nächstes die einzige vorhandene Zeile for value in values: und ersetze ihre ursprüngliche Erhöhung durch die folgende Validierung. Der Block zeigt diese vorhandene for-Zeile zur Orientierung; füge keine zweite Schleife hinzu. set(value) sammelt Feldnamen; isinstance prüft Typen; any(...) fragt, ob mindestens ein Feldwert ungültig ist. Die erste Bedingung verlangt ein Dictionary mit genau drei nichtleeren String-Feldern und der freigegebenen Auswahl. Prüfe den Katalog, bevor du versuchst, einen Standzähler zu erhöhen. Die for-Zeile bleibt mit vier Leerzeichen eingerückt, ihre Validierung mit acht:
for value in values:
if (not isinstance(value, dict)
or set(value) != {"event_id", "booth", "choice"}
or any(not isinstance(item, str) or not item for item in value.values())
or value["choice"] != "favourite"):
raise ValueError("Invalid vote schema")
if value["booth"] not in counts:
raise ValueError("Unknown booth: " + value["booth"])
Füge nun unmittelbar nach dieser Katalogprüfung die Identitätsbehandlung hinzu, weiterhin innerhalb der Schleife. Vergleiche bei einer bekannten ID den vollständigen dekodierten Wert. Ein anderer Inhalt muss einen Fehler unter Nennung dieser ID auslösen; den ersten oder letzten Wert auszuwählen würde das Ergebnis von der Durchlaufreihenfolge abhängig machen. Eine identische Wiederholung erreicht continue, das ohne weitere Erhöhung zur nächsten Eingabe geht. Eine erstmalige ID wird gespeichert und gezählt:
event_id = value["event_id"]
if event_id in seen:
if seen[event_id] != value:
raise ValueError("Conflicting event_id: " + event_id)
continue
seen[event_id] = dict(value)
counts[value["booth"]] += 1
dict(value) bewahrt den akzeptierten Wert als Dictionary. JSON-Leerraum und Feldreihenfolge sind keine fachliche Identität. Der Dictionary-Schlüssel gilt global über die Partitionen des ausgewählten Batches; er ist weder ein Standname noch ein Partitions-/Offset-Paar. Unterschiedliche IDs mit demselben Stand und derselben Auswahl müssen jeweils die Erhöhung für eine erstmalige ID erreichen.
Ersetze zuletzt die alte Rückgabe, die jeden Eingabewert verwendete. Gib die akzeptierten eindeutigen Werte und alle Katalogzähler zurück. Diese Zeile hat vier Leerzeichen, außerhalb der Schleife, aber innerhalb der Funktion:
return {"votes": list(seen.values()), "counts": counts}
Speichere und beende. In nanos unteren Beschriftungen bedeutet ^, Ctrl gedrückt zu halten: Drücke Ctrl+O, dann Enter, um reducer.py zu bestätigen; drücke Ctrl+X, um zur Shell zurückzukehren. Diese Tasten speichern die aktuelle Datei, statt Python auszuführen. Lies nach, was du tatsächlich gespeichert hast:
cat reducer.py
Prüfe, dass seen einmal je Funktionsaufruf initialisiert wird, Schema-/Katalogprüfungen vor dem Zählen stattfinden, nur beim ersten Auftreten einer ID gezählt wird und die Rückgabe außerhalb der Schleife steht. Teste danach sofort die fertige Funktion:
bigdata-python -B check_rules.py
Erwartet wird Rules pass für Identitäten, verschiedene IDs, Konflikte, Schema, Stände ohne Stimmen und umgeordnete Eingaben. Ein Dictionary nur mit dem Stand als Schlüssel würde verschiedene Stimmenereignisse zusammenfassen; eines mit Partition/Offset als Schlüssel würde alle Wiederholungen behalten. Dein Dictionary bildet stattdessen die fachliche Identität ab.
Lies bei einem Einrückungs-, Syntax- oder Namensfehler den gemeldeten Dateinamen und die Zeile, öffne erneut nano reducer.py und korrigiere den betroffenen Teil mit den Pfeiltasten. Speichere mit Ctrl+O, Enter, beende mit Ctrl+X, prüfe mit cat reducer.py und führe die Regelprüfung erneut aus. Bewahre die bereits richtigen Teile; du musst weder die ganze Funktion neu schreiben noch die VM zurücksetzen. Einen falschen Pfad findest du mit diesen nur lesenden Befehlen:
pwd
ls reducer.py replay.py
Kehre bei Bedarf mit cd /home/labex/project zurück. Der nächste Schritt verwendet den vorbereiteten nativen Leser, um deine Regel auf echte Kafka-Datensätze anzuwenden.
Veröffentliche und lies einen Snapshot aus dem echten Galerie-Log
In diesem Schritt wendest du deine Funktion auf den tatsächlichen eingefrorenen Galerie-Batch an. Bleibe in Terminal unter /home/labex/project. Liste die vom Batch main gewählte Quelle und Ausgabe auf und lies anschließend seinen Katalog:
bigdata-python -B vote_tool.py list
cat data/main-catalog.json
main bezeichnet gallery-votes. Es enthält mehrere Partitionen, wiederholte IDs an tatsächlich verschiedenen gespeicherten Positionen, unterschiedliche IDs mit sonst gleichem Inhalt und eine leere Partition. replay.py ist der vorbereitete native Leser; vote_support.py stellt Argumentverarbeitung und Snapshot-Schreiben bereit. Der Leser übergibt den Wert jeder tatsächlichen Datensatzhülle an deine Funktion summarize. Er bezieht die Werte nicht aus einer lokalen Stimmen-Eingabedatei.
Führe den folgenden öffentlichen Einstiegspunkt aus. Er prüft zuerst die aktuelle Identitätsfunktion mit wegwerfbaren Beispielen und führt dann dein aktuelles Programm mit seinen Helfer-Kindprozessen aus. Für diesen Prozessbaum bewahrt er Nachweise zum nativen Lesen und speichert das erfolgreiche Ergebnis im angegebenen Ausgabepfad:
bigdata-python -B run_replay.py main
Erwartet wird eine Veröffentlichung mit records=8, logical_votes=5. Der Einstiegspunkt erfasst, dass dieser Programmbaum tatsächlich Broker-Daten empfängt, statt einem fremden Consumer mit demselben Client-Label zu vertrauen. Er beweist nicht, dass die interne Berechnung tatsächlich auf diesen Daten beruht, und bietet keinen Schutz gegen Fälschungen durch Administratoren. Das endgültige fachliche Ergebnis wird separat mit der vollständigen tatsächlichen Quelle verglichen.
Jeder erfolgreiche Lauf muss das Ergebnis während dieses Aufrufs tatsächlich veröffentlichen: Dateidaten direkt schreiben oder eine temporäre Datei schreiben und zur Ergebnisdatei umbenennen. Auch ein Helfer-Kindprozess darf schreiben. Dieselben Bytes erneut zu veröffentlichen ist gültig; nur Kafka zu lesen und eine alte Datei zu behalten ist keine neue Veröffentlichung.
Lies die tatsächliche Ergebnisdatei. Dieser Leser lädt fehlende Ergebnisse niemals stillschweigend nach, ergänzt oder repariert sie:
bigdata-python -B read_snapshot.py output/main-snapshot.json
Erwartet werden fünf logische Stimmen-IDs und diese Standzahlen:
Booth Votes
fern 2
oak 3
reed 0
Der Snapshot bewahrt auch alle acht physischen Datensatzhüllen. So kannst du die Quelldatensätze von den fünf logischen Stimmen unterscheiden. reed bleibt mit null Stimmen enthalten. Beim vorbereiteten Schreibverfahren lassen Schleifen- oder Validierungsfehler vor der Veröffentlichung den vorherigen Snapshot verfügbar. Der Einstiegspunkt macht andere Schreibverfahren nicht automatisch rückgängig. Führe nach jeder Programmbearbeitung den öffentlichen Einstiegspunkt erneut aus, bevor du die Fertigstellung prüfst.
Die ursprüngliche Anzeigeanwendung ist pausiert und hat ihren Consumer-Gruppenfortschritt gespeichert. Du darfst diese Gruppe beschreiben, aber ihr nicht beitreten, sie nicht zurücksetzen/löschen und keine Offsets für sie bestätigen. Bearbeite/lösche das Quell-Topic nicht und füge keine Partitionen hinzu. Rekonstruiere nur deine eigene abgeleitete Ausgabe. Die Prüfung beobachtet die ursprüngliche Topic-Identität, die vollständige Partitions-/Replikat-Topologie, tatsächliche Datensätze und den gespeicherten Fortschritt der Anzeigeanwendung. Sie verspricht nicht, beliebige Broker-Konfigurationen zu schützen.
Du besitzt jetzt einen nützlichen Stand-Snapshot statt nur eines Zustellprotokolls. Das nächste Lab erklärt und erstellt den vollständigen unabhängigen nativen Leser, einschließlich der Frage, warum ein leerer Poll keinen abgeschlossenen Batch bedeutet.
Zusammenfassung
Du hast eine echte gespeicherte Wiederholung beobachtet, globale fachliche Identität von Kafka-Datensatzpositionen getrennt und widersprüchliche Inhalte mit derselben ID abgelehnt. Dein echter Galerie-Snapshot bewahrt die vollständige physische Quelle und zeigt zugleich logische Stimmen und jeden Stand, auch solche ohne Stimmen. Die Quelle und die Gruppe der pausierten Anzeigeanwendung bleiben verfügbar.
