Einführung
Ein Studio hat seine Stimmenhistorie aufbewahrt, während die ursprüngliche Anzeigeanwendung pausiert. Du brauchst eine neue Zählung, ohne den Fortschritt dieser Anwendung zurückzusetzen oder das Log zu verändern. Ein Poll kann kurzzeitig keine Nachricht liefern, obwohl ein neuer Consumer den eingefrorenen Batch noch nicht erhalten hat.
Du baust die native Replay-Schleife, erfasst für jede Partition getrennt die Endposition und veröffentlichst erst, wenn der vollständige endliche Batch gelesen wurde. Die Funktion für logische Identität aus dem vorherigen Lab ist in dieser unabhängigen VM bereits vorhanden. Du brauchst die dort verwendeten Python- und Kafka-Grundlagen. Verwende Terminal. Die Autorenschätzung von 50–60 Minuten enthält Lesen und gewöhnliche Fehler; sie ist kein mit Anfängern durchgeführter Zeittest.
Hier wird ein eingefrorener Batch rekonstruiert. Während der Übung werden keine weiteren Daten in diese Topics veröffentlicht, du musst keine Transaktion implementieren und es gibt keine Zusage für zuverlässige Wiederherstellung nach Abstürzen oder Ende-zu-Ende-Exactly-once-Verarbeitung.
Erfasse jede Partitionsgrenze und lies bis zu ihr
In diesem Schritt implementierst du einen vollständigen unabhängigen nativen Lesevorgang innerhalb eines vorbereiteten äußeren Programms. Beginne in Terminal. Bearbeite replay.py und bei Bedarf reducer.py oder Python-Helfer unter helpers/; schreibe abgeleitete Dateien unter output/. Bewahre die bereitgestellten Werkzeuge/Daten, ursprünglichen Topics und Gruppen der pausierten Anzeigeanwendungen:
cd /home/labex/project
bigdata-python -B vote_tool.py list
bigdata-python -B vote_tool.py describe main
Die nur lesende Beschreibung zeigt die tatsächlichen Partitions-IDs von studio-votes, den aufbewahrten Anfang low, das aktuelle eingefrorene Ende und die Replikate. Eine Partition ist ein geordnetes Log, keine separate VM. Das Ende ist die nächste Position nach dem aufbewahrten Batch, kein zu zählender Datensatz. Sage voraus, welche Partitionen überhaupt Datensätze benötigen: Eine leere Partition hat low == high und ist bereits vollständig.
Eine Offset-Spanne ist keine Datensatzanzahl. Kafka kann interne/Steuerpositionen oder entfernte Positionen überspringen. Diese Quelle enthält eine native übersprungene Position; du musst keine fehlende Stimme erfinden. Untersuche die echten Datensatzhüllen und vergleiche ihre tatsächlichen Positionen mit der Beschreibung:
bigdata-python -B vote_tool.py inspect main
Lies vor der Rekonstruktion die gespeicherten Positionen der pausierten Anzeigeanwendung mit dem Consumer-Gruppenwerkzeug aus dem vorausgesetzten Kurs:
kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group studio-votes-display --describe
Bewahre jeden CURRENT-OFFSET für den späteren Vergleich. Keine aktiven Mitglieder sind normal: Die Anzeigeanwendung pausiert, hat aber gespeicherten Fortschritt. LAG ist hier eine Differenz zwischen Logpositionen, keine Anzahl logischer Stimmen. Dein unabhängiger Consumer darf den Fortschritt dieser Gruppe nicht verändern.
Lies das vorbereitete äußere Programm
cat replay.py
Imports, arguments(), Clientkonfiguration, Fehlermeldungen und finally/Schließen sind bereits vorhanden. Der Kern löst absichtlich NotImplementedError aus; er liest keine Partitionen und veröffentlicht keine Musterantwort. Die richtige Identitätsfunktion aus dem vorherigen Lab ist getrennt in reducer.py vorbereitet.
Der Client verwendet die ausgewählte Adresse bootstrap.servers und eine unabhängige group.id. allow.auto.create.topics ist false. Automatischer Commit und automatische Offset-Speicherung sind deaktiviert. assign wird ihm explizite Startpositionen ohne dynamische Gruppenzuteilung geben; dieses Beispiel führt keinen expliziten Commit-Aufruf aus. Eine gleichwertige Implementierung mit unabhängiger Gruppe darf ihren eigenen Fortschritt speichern, aber niemals einer ursprünglichen Anzeigegruppe beitreten, für sie Offsets bestätigen, sie zurücksetzen oder löschen.
Das vorbereitete vote_support.py stellt arguments() für Topic/Katalog/Ausgabe/Broker/Gruppe und finish(...) zum Schreiben vollständiger Ergebnisse bereit. Es erkennt keine Partitionen und ruft keine Stimmen ab. finish ruft vor dem Schreiben/Ersetzen des Ergebnisses die Identitätsregel auf. finally schließt den Consumer nach Erfolg oder Fehler; bewahre diese äußere Struktur.
Implementiere den nativen Kern in kurzen zusammenhängenden Teilen
Verwende den Editorablauf aus dem vorherigen Lab:
nano replay.py
Erhalte Imports, Argumentverarbeitung, Consumer-Konfiguration, try:, except und finally. Entferne innerhalb von try: nur die beiden erklärenden Kommentare und das raise NotImplementedError(...) des unvollständigen Kerns. Positioniere dich mit den Pfeiltasten und bearbeite mit Rücktaste/Entf; Ctrl+K kann die gesamte aktuelle Zeile ausschneiden, verwende es daher nur auf einer Zeile, die du ersetzen möchtest.
Füge die folgenden zusammenhängenden Teile in diese Lücke ein. Sie bilden einen Kern: Führe einzelne Blöcke nicht an der Shell aus und speichere/starte keine halbfertige Schleife. Vervollständige alle Teile, bewahre das bereitgestellte äußere Fehler-/Schließgerüst, speichere dann und führe aus. Vier Leerzeichen setzen eine Anweisung in try:; weitere vier setzen sie in eine Schleife oder Bedingung. Die gezeigte relative Einrückung ist Teil von Pythons Bedeutung.
Erkenne zuerst die vollständige tatsächliche Partitionsmenge. list_topics ohne angefordertes Topic aufzurufen vermeidet eine Metadatenanfrage für einen unbekannten Namen. Prüfe vor der Verwendung des ausgewählten Eintrags seine Verfügbarkeit. Das Sortieren der erkannten IDs gibt diesem Beispiel eine stabile Zuweisungsliste; es erzeugt keine globale Ereignisreihenfolge:
metadata = consumer.list_topics(timeout=10)
if args.topic not in metadata.topics or metadata.topics[args.topic].error:
raise ValueError("The selected topic is unavailable")
partitions = sorted(metadata.topics[args.topic].partitions)
Fordere für jede erkannte Partition einmal get_watermark_offsets(..., timeout=10) an. TopicPartition identifiziert Topic und Partition für diese Anfrage. Bewahre sowohl den aufbewahrten Anfang low als auch das eingefrorene Ende high und initialisiere die nächste Diagnoseposition mit low. Aktualisiere high nicht innerhalb der Polling-Schleife: Einem wachsenden Ende zu folgen wäre keine endliche Neuzählung mehr. Diese Übung verbietet, die eingefrorene Quelle zu erweitern oder zu verändern:
bounds = {}
positions = {}
for number in partitions:
low, high = consumer.get_watermark_offsets(
TopicPartition(args.topic, number), timeout=10)
bounds[number] = [low, high]
positions[number] = low
Erzeuge pending nur aus Partitionen mit noch zu lesenden aufbewahrten Positionen. assign erhält eine Liste expliziter Start-Offsets, einschließlich der Anfangswerte leerer Partitionen. Initialisiere für jeden Lauf eine frische Liste physischer Datensätze. time.monotonic() misst verstrichene Zeit unabhängig von Änderungen der Systemuhr; die Frist begrenzt einen unvollständigen Lesevorgang:
pending = {number for number in partitions
if bounds[number][0] < bounds[number][1]}
consumer.assign([TopicPartition(args.topic, number, bounds[number][0])
for number in partitions])
records = []
deadline = time.monotonic() + 30
Beginne nun die Schleife. poll(0.2) kann eine Nachricht, eine Nachricht mit Fehler/Ereignis oder bei Timeout None zurückgeben. Sage voraus, warum None nicht fertig bedeuten kann: Der Client hat einen eingefrorenen Datensatz vielleicht einfach noch nicht erhalten. Warte innerhalb der gesamten Frist weiter. Partitions-EOF-Ereignisse sind in der Konfiguration aktiviert; _PARTITION_EOF ist das erwartete Endereignis, andere Fehler lassen den Lauf scheitern. Dies ermöglicht nativen Fortschritt über eine letzte interne Position hinaus:
while pending:
if time.monotonic() >= deadline:
raise RuntimeError("Incomplete replay: frozen ends were not reached")
message = consumer.poll(0.2)
if message is not None:
if message.error():
if message.error().code() != KafkaError._PARTITION_EOF:
raise RuntimeError(str(message.error()))
Setze denselben Block if message is not None mit dem folgenden elif fort, das mit zwölf Leerzeichen an if message.error(): ausgerichtet ist. Nur eine normale Nachricht in einer offenen Partition unterhalb ihres festen high wird zur Datensatzhülle. Bewahre auch physische Wiederholungen; summarize behandelt die logische Identität später. Dekodiere UTF-8-Bytes von Schlüssel und Wert; json.loads verwandelt den Wert dann in ein Dictionary:
elif (message.partition() in pending
and message.offset() < bounds[message.partition()][1]):
records.append({
"topic": message.topic(),
"partition": message.partition(),
"offset": message.offset(),
"key": message.key().decode("utf-8"),
"value": json.loads(message.value().decode("utf-8")),
})
Frage nach der Behandlung dieses Polls consumer.position(...) für jede verbleibende Partition ab. Dies ist die nächste Position des Consumers, getrennt vom bestätigten Fortschritt irgendeiner Gruppe. Vergleiche sie jeweils mit dem erfassten high, nicht mit einer geschätzten Datensatzanzahl. Entferne nur vollständige Partitionen aus pending. Die Zeile for position hat acht Leerzeichen, innerhalb von while, aber außerhalb der Nachrichtenbedingung, sodass auch auf einen inaktiven Poll eine Positionsprüfung folgen kann. Rufe zuletzt finish mit vier Leerzeichen auf, nachdem die Schleife abgeschlossen ist:
for position in consumer.position(
[TopicPartition(args.topic, number) for number in pending]):
positions[position.partition] = position.offset
if position.offset >= bounds[position.partition][1]:
pending.remove(position.partition)
finish(args, records, bounds, positions)
Jede Partition hat nun eine feste Grenze. Leere Partitionen benötigten keine Stimme; übersprungene Positionen erforderten keine erfundenen Datensätze; None beendete das Lesen nicht. Erst nachdem alle offenen nächsten Positionen ihre Enden erreicht haben, wird ein vollständiger Snapshot veröffentlicht.
Speichere mit Ctrl+O, Enter, um replay.py zu bestätigen, dann Ctrl+X. Prüfe das gespeicherte Programm:
cat replay.py
Prüfe, dass alle sechs Teile innerhalb des bereitgestellten try liegen, die Behandlung der Datensatzhüllen im Zweig für normale Nachrichten bleibt, die Positionsprüfung innerhalb der Schleife bleibt und finish außerhalb steht. Führe es sofort für den ausgewählten nativen Batch aus:
bigdata-python -B run_replay.py main
bigdata-python -B read_snapshot.py output/main-snapshot.json
Erwartet werden acht physische Datensätze, fünf logische Stimmen und cedar=3, maple=2, willow=0. Der öffentliche Einstiegspunkt prüft zuerst die Identitätsfunktion mit wegwerfbaren Beispielen und erfasst dann Broker-Daten, die der tatsächliche native Programm-/Helfer-Prozessbaum empfängt. Ein separater vollständiger fachlicher Vergleich prüft das tatsächliche Ergebnis. Vom Snapshot angegebene Grenzen/Positionen dienen der Diagnose, sind kein unabhängiger Nachweis nativer Verarbeitung.
Jeder erfolgreiche Aufruf muss sein Ergebnis tatsächlich schreiben, direkt oder durch eine geschriebene temporäre Datei, die zur ausgewählten Ausgabe umbenannt wird. Identische Bytes bei Wiederholung sind gültig, und Helfer-Kindprozesse dürfen schreiben. Eine alte Datei nur zu behalten oder ihren Zeitstempel zu aktualisieren oder eine Fertigmarkierung zu schreiben ist keine Veröffentlichung. Das bereitgestellte finish verwendet das gelehrte Verfahren mit temporärer Datei und Ersetzung. Bei diesem vorbereiteten Schreibverfahren lassen Schleifen- oder Validierungsfehler vor der Veröffentlichung den früheren Snapshot verfügbar; der Einstiegspunkt macht beliebige direkte Schreibzugriffe nicht rückgängig.
Öffne bei einem Syntax-, Namens- oder Einrückungsfehler anhand der gemeldeten Zeile erneut nano replay.py, korrigiere diesen Teil und bewahre die funktionierenden Teile. Speichere/beende, prüfe mit cat replay.py und starte den öffentlichen Einstiegspunkt erneut. Prüfe bei einem Timeout wegen unvollständigen Replays das ausgewählte Topic-Argument sowie die Behandlung von pending, EOF und Positionen; dies ist kein Grund, die ursprüngliche Gruppe zurückzusetzen. Verwende pwd und ls replay.py reducer.py, um einen falsch geschriebenen Pfad zu finden. Ein lokaler Programmfehler erfordert weder erneutes Setup noch eine Änderung der Quelldatensätze.
Rekonstruiere von Grund auf und verwende einen anderen Batch
In diesem Schritt weist du nach, dass ein weiterer Lauf dasselbe endliche Ergebnis rekonstruiert, und verwendest dann einen neuen Katalog mit einer anderen Partitionsverteilung. Bleibe in Terminal unter /home/labex/project.
Das vorbereitete Schreibverfahren sortiert physische Datensatzhüllen vor dem JSON-Speichern nach Partition/Offset und logische Stimmen nach ID. Dadurch bleibt derselbe vollständige Snapshot stabil, ohne eine globale Ereignisreihenfolge zu behaupten. sha256sum berechnet einen Fingerabdruck des Inhalts. > speichert diesen unter deinem Ausgabeverzeichnis; -c vergleicht später die tatsächlich benannte Datei mit dem gespeicherten Fingerabdruck:
sha256sum output/main-snapshot.json > output/main.sha256
bigdata-python -B run_replay.py main
sha256sum -c output/main.sha256
Erwartet wird output/main-snapshot.json: OK. Bei jedem Aufruf beginnt das Programm mit leeren records, seen und Zählern. Es ersetzt einen vollständigen Snapshot, statt die historischen Stimmen erneut auf die bisherigen Summen zu addieren. Der Konsum hat diese Datensätze nicht gelöscht und die pausierte Anzeigeanwendung nicht zurückgespult.
Lies die Positionen der ursprünglichen Anzeigeanwendung nach der Wiederholung noch einmal:
kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group studio-votes-display --describe
Jeder CURRENT-OFFSET muss mit der zuvor aufgenommenen Tabelle der pausierten Anwendung übereinstimmen. Deine unabhängige Rekonstruktion hat für diese Gruppe weder Offsets bestätigt noch ihren Fortschritt zurückgesetzt, obwohl sie die aufbewahrten Datensätze gelesen hat. Die Meldung über fehlende aktive Mitglieder bleibt normal.
Liste den Batch check auf und lies seinen eigenen Standkatalog:
bigdata-python -B vote_tool.py list
cat data/check-catalog.json
Dieser Batch hat andere Standnamen und drei statt vier Partitionen. Behalte dieselben Programmdateien bei. Die Argumentverarbeitung wählt sein tatsächliches Topic und seinen Katalog; die Erkennung findet die vollständige reale Partitionsmenge:
bigdata-python -B run_replay.py check
bigdata-python -B read_snapshot.py output/check-snapshot.json
Erwartet werden sechs physische Datensätze, vier logische Stimmen und amber=2, violet=2, moss=0. Den ersten Katalog fest einzuprogrammieren oder nach der Datensatzanzahl des ersten Topics aufzuhören würde diesen Batch nicht korrekt beschreiben.
Kehre zur ersten tatsächlichen Datei zurück:
bigdata-python -B read_snapshot.py output/main-snapshot.json
Sie enthält weiterhin das Ergebnis des Studios. Der Leser öffnet die angeforderte Datei direkt und ergänzt nichts aus Kafka. Meldet er eine fehlende Datei, vergleiche den Dateinamen mit vote_tool.py list und korrigiere den Pfad. Rekonstruiere den zugehörigen Batch nur, wenn dessen Ergebnis tatsächlich neu aufgebaut werden muss.
Lehne einen Konflikt ab und bewahre den gültigen Snapshot
In diesem Schritt beobachtest du einen Datenfehler, ohne ein nützliches Ergebnis zu ersetzen. Bleibe in Terminal unter /home/labex/project. Das aufgelistete Topic conflict enthält absichtlich eine ID mit unterschiedlichen, jeweils gültigen Standinhalten. Sein Ausgabeziel ist der vorhandene Haupt-Snapshot. Kein Wert ist als Gewinner freigegeben: Dieser Lauf darf kein Ergebnis als gültig veröffentlichen.
Der Fingerabdruck aus dem vorherigen Schritt beschreibt die gültige Hauptdatei. Führe dasselbe Programm für den Konflikt-Batch aus. Der nächste Befehl soll fehlschlagen:
bigdata-python -B run_replay.py conflict
printf '%s\n' "$?"
Erwartet werden ein Exitstatus ungleich null und Conflicting event_id mit der tatsächlichen ID. Die Regel läuft, bevor das vorbereitete Schreibverfahren das Ergebnis erstellt/ersetzt. Das Verfahren mit .pending und Ersetzung ist hier eine Technik für die Veröffentlichung vollständiger Ergebnisse, keine Garantie für Absturzwiederherstellung oder Dauerhaftigkeit.
sha256sum -c output/main.sha256
bigdata-python -B read_snapshot.py output/main-snapshot.json
Die gültige Hauptdatei muss weiterhin zum Fingerabdruck passen und die ursprünglichen Studiozahlen enthalten. Der fehlgeschlagene Konfliktlauf wird getrennt von erfolgreichen main/check-Läufen aufgezeichnet. Bearbeite die Konfliktquelle nicht und wähle nicht einfach das zuerst eingetroffene Ereignis aus. Auch ein umgekehrter oder sequenzieller Partitionsdurchlauf muss denselben Konflikt ablehnen.
Bei einem gewöhnlich falsch geschriebenen Batchnamen zeigt der Einstiegspunkt vor Beginn des Replays die verfügbaren Namen. Verwende vote_tool.py list und starte erneut mit dem richtigen Namen. Behebe bei einem lokalen Programmfehler die Datei und starte den betroffenen gültigen Batch erneut. Nach jeder Quellcodeänderung musst du beide gültigen Batches wieder ausführen, damit ihre Ergebnisse zu den aktuellen Programmdateien gehören; wiederhole anschließend die Konflikt-Selbstprüfung.
Lasse die ursprünglichen Anzeigegruppen pausiert: Beitritt, Commit, Zurücksetzen oder Löschen sind verboten. Eine Gruppe zurückzusetzen und danach denselben finalen Offset wiederherzustellen liegt ebenfalls außerhalb eines unabhängigen Replays. Die Funktionsprüfungen verbinden nativen Empfang durch den Programmbaum mit tatsächlichen vollständigen Ausgaben. Sie beweisen nicht, dass die Berechnung tatsächlich auf diesen Daten beruht, und bieten keinen Schutz vor einem Administrator, der die Umgebung fälscht.
Zusammenfassung
Du hast reale Grenzen pro Partition erfasst und die vollständigen endlichen Logs trotz kurzzeitig inaktiver Polls, leerer Partitionen und übersprungener Positionen gelesen. Dasselbe native Programm hat zwei Kataloge rekonstruiert, ohne alte Summen anzuhäufen. Eine widersprüchliche ID ließ den Lauf fehlschlagen, während der nützliche Haupt-Snapshot und der ursprüngliche Consumer-Fortschritt erhalten blieben.
