Der Verkauf, der zu spät kam
Ein Zeitfenster zählt Ereignisse in einem festen Abschnitt der Zeit, zum Beispiel pro Minute. Dabei kommt es darauf an, welche Uhr gilt: die Zeit, zu der etwas geschah, oder die, zu der es ankam.
Eine Kette zählt jede Minute, wie viele Artikel ihre Kassen verkaufen. Eine Kasse verliert für ein paar Sekunden das Netz. Danach sendet sie einen Verkauf von 10:00:58 nach einem Verkauf von 10:01:02. In welche Minute gehört dieser Verkauf? Heute zählen Sie auf beide Arten und sehen den Unterschied.
Am Ende dieser Lektion
Notiz
Ich kann Ereignisse in festen Fenstern der Ereigniszeit zählen und sagen, welche verspätet sind.
Schlüsselwörter
| Begriff | Was er bedeutet |
|---|---|
| Ereigniszeit | der Moment, in dem es wirklich geschah (Event Time) |
| Verarbeitungszeit | der Moment, in dem das System das Ereignis erhielt (Processing Time) |
| Festes Fenster | ein Zeitabschnitt, der das nächste nie überlappt (Tumbling Window) |
| Verspätetes Ereignis | ein Ereignis, das nach einem späteren eintrifft |
| Erlaubte Verspätung | wie lange ein Fenster auf verspätete Ereignisse wartet |
Die Zellen, die Lernende ausfüllen, bleiben in dieser Vorschau leer.
Jedes Ereignis hat zwei Uhren
Die Ereigniszeit ist der Moment, in dem etwas geschah, etwa der Augenblick eines Verkaufs. Die Verarbeitungszeit ist der Moment, in dem Ihr System das Ereignis erhielt.
Meist liegen sie nah beieinander. Ein Netzausfall, ein Handy ohne Empfang oder ein ausgelasteter Server können sie auseinanderziehen. Kafka hält die Ereignisse in der Reihenfolge des Eintreffens: Seine Offsets folgen der Verarbeitungszeit, nicht der Ereigniszeit.
Beispiel: fünf Verkäufe
| Offset | Artikel | Ereigniszeit (Sekunden) | Eingetroffen |
|---|---|---|---|
| 0 | brot | 5 | als erstes |
| 1 | milch | 62 | als zweites |
| 2 | kaffee | 58 | als drittes, nach 62 |
| 3 | reis | 125 | als viertes |
| 4 | eier | 70 | als fünftes, nach 125 |
Die Zellen, die Lernende ausfüllen, bleiben in dieser Vorschau leer.
Ein Topic mit einer Partition anlegen
curl -s -X POST -H "Content-Type: application/json" -d '{"name":"kassen","partitions":1,"replicationFactor":1,"configs":{}}' http://kafka-ui:8080/api/clusters/lab/topics -o /dev/null -w "%{http_code}\n"Hier angezeigt, nicht ausgeführt. Erstellen Sie ein kostenloses Konto, um „Ausführen“ zu drücken und ihn im Terminal Ihres eigenen Labs einzugeben.
Die fünf Verkäufe in Ankunftsreihenfolge senden
for e in "5 brot" "62 milch" "58 kaffee" "125 reis" "70 eier"; do curl -s -X POST -H "Content-Type: application/json" -d "{\"partition\":0,\"key\":\"kasse\",\"value\":\"$e\",\"headers\":{},\"keySerde\":\"String\",\"valueSerde\":\"String\"}" http://kafka-ui:8080/api/clusters/lab/topics/kassen/messages -w "%{http_code}\n"; doneHier angezeigt, nicht ausgeführt. Erstellen Sie ein kostenloses Konto, um „Ausführen“ zu drücken und ihn im Terminal Ihres eigenen Labs einzugeben.
Die Offsets ansehen
curl -s http://kafka-ui:8080/api/clusters/lab/topics/kassen | grep -o '"partition":[0-9]*\|"offsetMax":[0-9]*'Hier angezeigt, nicht ausgeführt. Erstellen Sie ein kostenloses Konto, um „Ausführen“ zu drücken und ihn im Terminal Ihres eigenen Labs einzugeben.
Das Topic von Anfang an lesen
curl -s -N "http://kafka-ui:8080/api/clusters/lab/topics/kassen/messages/v2?seekType=BEGINNING&limit=10" | grep -o '"offset":[0-9]*\|"value":"[^"]*"'Hier angezeigt, nicht ausgeführt. Erstellen Sie ein kostenloses Konto, um „Ausführen“ zu drücken und ihn im Terminal Ihres eigenen Labs einzugeben.
Die Uhrminuten des Brokers zählen
curl -s -N "http://kafka-ui:8080/api/clusters/lab/topics/kassen/messages/v2?seekType=BEGINNING&limit=10" | grep -o '"timestamp":"[^"]*"' | cut -c25-29 | sort -u | wc -lHier angezeigt, nicht ausgeführt. Erstellen Sie ein kostenloses Konto, um „Ausführen“ zu drücken und ihn im Terminal Ihres eigenen Labs einzugeben.
Ein Fenster ist ein Stück Ereigniszeit
Um pro Minute zu zählen, schneiden Sie die Zeit in Fenster zu 60 Sekunden. Ein festes Fenster beginnt dort, wo das vorige endet: Jedes Ereignis liegt in genau einem.
Der Anfang des Fensters eines Ereignisses ist seine Zeit minus der Rest der Division durch die Größe. Die Zeit 62 ergibt 62 minus 2: Das Fenster beginnt bei 60.
Beispiel: Fenster zu 60 Sekunden
| Fenster | Anfang | Ende (ausgeschlossen) | Artikel |
|---|---|---|---|
| Erstes | 0 | 60 | brot, kaffee |
| Zweites | 60 | 120 | milch, eier |
| Drittes | 120 | 180 | reis |
Die Zellen, die Lernende ausfüllen, bleiben in dieser Vorschau leer.
Die Ereignisse jedes Fensters zählen
# Jedes Ereignis ist [Ereigniszeit in Sekunden, Artikel], in der Reihenfolge des Eintreffens.
ereignisse = [[5, "brot"], [62, "milch"], [58, "kaffee"], [125, "reis"], [70, "eier"]]
groesse = 60
zaehler = {}
for ereignis in ereignisse:
start = ereignis[0] // groesse * groesse
if start in zaehler:
zaehler[start] = zaehler[start] + 1
else:
zaehler[start] = 1
print(zaehler)
# Setzen Sie groesse auf 30 und führen Sie es erneut aus.
Die Fenster aus dem Topic selbst zählen
curl -s -N "http://kafka-ui:8080/api/clusters/lab/topics/kassen/messages/v2?seekType=BEGINNING&limit=10" | grep -o '"value":"[^"]*"' | cut -d'"' -f4 | while read t artikel; do echo $((t / 60)); done | sort -n | uniq -cHier angezeigt, nicht ausgeführt. Erstellen Sie ein kostenloses Konto, um „Ausführen“ zu drücken und ihn im Terminal Ihres eigenen Labs einzugeben.
Ein häufiger Fehler
Warnung
Viele glauben, die Reihenfolge im Topic sei die Reihenfolge der Ereignisse. Es ist die Reihenfolge des Eintreffens. Wählen Sie das Fenster mit der Zeit im Ereignis, und nutzen Sie Offsets nur, um zu sagen, was Sie gelesen haben.
Ein verspätetes Ereignis kommt nach einem späteren
Der Kaffee wurde bei 58 Sekunden verkauft, kam aber nach der Milch von 62 an. Das ist ein verspätetes Ereignis: Ein späteres kam zuerst.
Ein echtes System kann nicht ewig warten. Es schließt ein Fenster, sobald die neueste gesehene Zeit dessen Ende überschreitet. Eine kleine erlaubte Verspätung hält das Fenster etwas länger offen.
Die verspäteten Ereignisse finden
# Ein Ereignis ist verspätet, wenn vor ihm ein Ereignis mit späterer Zeit eingetroffen ist.
ereignisse = [[5, "brot"], [62, "milch"], [58, "kaffee"], [125, "reis"], [70, "eier"]]
neueste = 0
for ereignis in ereignisse:
if ereignis[0] < neueste:
print(ereignis[1], "ist verspätet: seine Zeit", ereignis[0], "liegt vor", neueste)
else:
neueste = ereignis[0]
# Hängen Sie [30, "tee"] an ereignisse an und führen Sie es erneut aus.
Ein Fenster nach der erlaubten Verspätung schließen
# Ein Fenster schließt, sobald die neueste Zeit sein Ende plus die erlaubte Verspätung erreicht.
ereignisse = [[5, "brot"], [62, "milch"], [58, "kaffee"], [125, "reis"], [70, "eier"]]
groesse = 60
verspaetung = 5
neueste = 0
zaehler = {}
verworfen = []
for ereignis in ereignisse:
start = ereignis[0] // groesse * groesse
if neueste >= start + groesse + verspaetung:
verworfen.append(ereignis[1])
elif start in zaehler:
zaehler[start] = zaehler[start] + 1
else:
zaehler[start] = 1
if ereignis[0] > neueste:
neueste = ereignis[0]
print(zaehler, verworfen)
# Setzen Sie verspaetung auf 0, dann auf 10, und führen Sie es erneut aus.
Noch einen sehr späten Verkauf senden
curl -s -X POST -H "Content-Type: application/json" -d "{\"partition\":0,\"key\":\"kasse\",\"value\":\"30 tee\",\"headers\":{},\"keySerde\":\"String\",\"valueSerde\":\"String\"}" http://kafka-ui:8080/api/clusters/lab/topics/kassen/messages -w "%{http_code}\n"Hier angezeigt, nicht ausgeführt. Erstellen Sie ein kostenloses Konto, um „Ausführen“ zu drücken und ihn im Terminal Ihres eigenen Labs einzugeben.
Die Fenster noch einmal zählen
curl -s -N "http://kafka-ui:8080/api/clusters/lab/topics/kassen/messages/v2?seekType=BEGINNING&limit=10" | grep -o '"value":"[^"]*"' | cut -d'"' -f4 | while read t artikel; do echo $((t / 60)); done | sort -n | uniq -cHier angezeigt, nicht ausgeführt. Erstellen Sie ein kostenloses Konto, um „Ausführen“ zu drücken und ihn im Terminal Ihres eigenen Labs einzugeben.
Zum Weiterdenken: eine Verspätung wählen
Führen Sie den Code mit erlaubter Verspätung 0, 5 und 10 aus. Notieren Sie, wie viele Ereignisse jede verwirft.
Antworten Sie dann in einem Satz: ein Kassenbericht für die Filialleiterin am Tagesende und ein Alarm, der in Sekunden auslösen muss. Welcher kann auf verspätete Ereignisse warten, und warum?