The sale that arrived late
A window over event time counts events by when they really happened, such as the sales made in each minute. It matters because events often arrive late, and counting them by arrival puts them in the wrong minute.
A shop counts how many items its tills sell every minute. A till goes offline for a few seconds, then sends a sale made at 10:00:58 after one made at 10:01:02. Which minute does that sale belong to: the minute it was made, or the minute it arrived?
By the end of this lesson
Note
I can count events in fixed windows of event time and say which events are late.
Key words
| Word | What it means |
|---|---|
| Event time | when the thing really happened |
| Processing time | when the system received the event |
| Tumbling window | a fixed slice of time that never overlaps the next |
| Late event | an event that arrives after a later one |
| Allowed lateness | how long a window waits for late events |
The cells students fill in are left blank in this preview.
Every event has two clocks
The event time is when the thing happened, such as the moment of a sale. The processing time is when your system got the event.
Most of the time they are close. A network fault, a phone with no signal or a busy server can make them far apart. Kafka keeps events in the order they arrive, so its offsets follow processing time, not event time.
Worked example: five sales
| Offset | Item | Event time (seconds) | Arrived |
|---|---|---|---|
| 0 | pasta | 5 | first |
| 1 | bread | 62 | second |
| 2 | milk | 58 | third, after 62 |
| 3 | rice | 125 | fourth |
| 4 | eggs | 70 | fifth, after 125 |
The cells students fill in are left blank in this preview.
Make a topic with one partition
curl -s -X POST -H "Content-Type: application/json" -d '{"name":"tills","partitions":1,"replicationFactor":1,"configs":{}}' http://kafka-ui:8080/api/clusters/lab/topics -o /dev/null -w "%{http_code}\n"Shown here, not run. Create a free account to press Run and type it into your own lab’s terminal.
Send the five sales in arrival order
for e in "5 pasta" "62 bread" "58 milk" "125 rice" "70 eggs"; do curl -s -X POST -H "Content-Type: application/json" -d "{\"partition\":0,\"key\":\"till\",\"value\":\"$e\",\"headers\":{},\"keySerde\":\"String\",\"valueSerde\":\"String\"}" http://kafka-ui:8080/api/clusters/lab/topics/tills/messages -w "%{http_code}\n"; doneShown here, not run. Create a free account to press Run and type it into your own lab’s terminal.
Look at the offsets
curl -s http://kafka-ui:8080/api/clusters/lab/topics/tills | grep -o '"partition":[0-9]*\|"offsetMax":[0-9]*'Shown here, not run. Create a free account to press Run and type it into your own lab’s terminal.
Read the topic from the beginning
curl -s -N "http://kafka-ui:8080/api/clusters/lab/topics/tills/messages/v2?seekType=BEGINNING&limit=10" | grep -o '"offset":[0-9]*\|"value":"[^"]*"'Shown here, not run. Create a free account to press Run and type it into your own lab’s terminal.
Count the clock minutes the broker stamped
curl -s -N "http://kafka-ui:8080/api/clusters/lab/topics/tills/messages/v2?seekType=BEGINNING&limit=10" | grep -o '"timestamp":"[^"]*"' | cut -c25-29 | sort -u | wc -lShown here, not run. Create a free account to press Run and type it into your own lab’s terminal.
A window is a slice of event time
To count per minute you cut time into windows of 60 seconds. A tumbling window starts where the last one ended, so each event is in exactly one.
The window of an event starts at its time minus what is left after dividing by the size. Time 62 gives 62 minus 2, so the window starts at 60.
Worked example: windows of 60 seconds
| Window | Starts | Ends before | Items |
|---|---|---|---|
| First | 0 | 60 | pasta, milk |
| Second | 60 | 120 | bread, eggs |
| Third | 120 | 180 | rice |
The cells students fill in are left blank in this preview.
Count the events in each window
# Each event is [event time in seconds, item], in the order it arrived.
events = [[5, "pasta"], [62, "bread"], [58, "milk"], [125, "rice"], [70, "eggs"]]
size = 60
counts = {}
for event in events:
start = event[0] // size * size
if start in counts:
counts[start] = counts[start] + 1
else:
counts[start] = 1
print(counts)
# Change size to 30 and run it again.
Count the windows from the topic itself
curl -s -N "http://kafka-ui:8080/api/clusters/lab/topics/tills/messages/v2?seekType=BEGINNING&limit=10" | grep -o '"value":"[^"]*"' | cut -d'"' -f4 | while read t item; do echo $((t / 60)); done | sort -n | uniq -cShown here, not run. Create a free account to press Run and type it into your own lab’s terminal.
A common mistake
Warning
Many people think the order in the topic is the order things happened. It is the order they arrived. Use the time inside each event to choose its window, and offsets only to say what you have read.
A late event arrives after a later one
The milk was sold at 58 seconds but arrived after the bread sold at 62. It is a late event: something with a later time got there first.
A real system cannot wait for ever. It closes a window once the newest time it has seen reaches the window's end. A small allowed lateness keeps the window open a little longer.
Find the late events
# An event is late when an event with a later time arrived before it.
events = [[5, "pasta"], [62, "bread"], [58, "milk"], [125, "rice"], [70, "eggs"]]
newest = 0
for event in events:
if event[0] < newest:
print(event[1], "is late: its time", event[0], "is before", newest)
else:
newest = event[0]
# Add [30, "tea"] at the end of events and run it again.
Close a window after an allowed lateness
# A window closes once the newest time seen reaches its end plus the allowed lateness.
events = [[5, "pasta"], [62, "bread"], [58, "milk"], [125, "rice"], [70, "eggs"]]
size = 60
lateness = 5
newest = 0
counts = {}
dropped = []
for event in events:
start = event[0] // size * size
if newest >= start + size + lateness:
dropped.append(event[1])
elif start in counts:
counts[start] = counts[start] + 1
else:
counts[start] = 1
if event[0] > newest:
newest = event[0]
print(counts, dropped)
# Change lateness to 0, then to 10, and run it again.
Send one more, a very late sale
curl -s -X POST -H "Content-Type: application/json" -d "{\"partition\":0,\"key\":\"till\",\"value\":\"30 tea\",\"headers\":{},\"keySerde\":\"String\",\"valueSerde\":\"String\"}" http://kafka-ui:8080/api/clusters/lab/topics/tills/messages -w "%{http_code}\n"Shown here, not run. Create a free account to press Run and type it into your own lab’s terminal.
Count the windows again
curl -s -N "http://kafka-ui:8080/api/clusters/lab/topics/tills/messages/v2?seekType=BEGINNING&limit=10" | grep -o '"value":"[^"]*"' | cut -d'"' -f4 | while read t item; do echo $((t / 60)); done | sort -n | uniq -cShown here, not run. Create a free account to press Run and type it into your own lab’s terminal.
Going further: choose a lateness
Run the lateness code again with a lateness of 0, 5 and 10. Write down how many events each one drops.
Then answer in a sentence: a till report for the manager at the end of the day, and an alarm that must fire in seconds. Which one can wait for late events, and why?