遅れて届いた売上
イベント時刻ごとの窓とは、イベントを、起きた時刻で一定の長さの時間に区切って数える方法です。届くのが遅れても、正しい区切りに入れられます。
あるチェーン店は、レジで売れた商品の数を1分ごとに数えています。あるレジが数秒だけ通信できなくなりました。
つながったあと、10:01:02 の売上のあとに、10:00:58 の売上を送ります。この売上は、どの1分に入れるべきでしょうか。起きた時刻でしょうか。届いた時刻でしょうか。今日は両方で数えて、違いを確かめます。
このレッスンのゴール
メモ
イベントをイベント時刻の固定の窓で数え、どれが遅れて届いたか言えます。
キーワード
| ことば | いみ |
|---|---|
| イベント時刻 | 実際に出来事が起きた時刻 |
| 処理時刻 | システムがイベントを受け取った時刻 |
| 固定の窓 | 次の窓と重ならない、一定の長さの時間の区切り |
| 遅れて届いたイベント | より新しいイベントのあとに届いたイベント |
| 許容遅延 | 窓が、遅れて届くイベントを待つ時間 |
学生が入力するセルは、このプレビューでは空白です。
イベントには時計が2つある
イベント時刻は、出来事が起きた時刻です。たとえば、売れた瞬間です。処理時刻は、あなたのシステムがイベントを受け取った時刻です。
ふつうは近い値です。通信の障害、電波のない携帯、混み合ったサーバーがあると、離れます。Kafkaは届いた順に並べます。オフセットは処理時刻の順で、イベント時刻の順ではありません。
例: 5件の売上
| オフセット | 商品 | イベント時刻（秒） | 届いた順 |
|---|---|---|---|
| 0 | bread | 5 | 1番目 |
| 1 | milk | 62 | 2番目 |
| 2 | coffee | 58 | 3番目（62のあと） |
| 3 | rice | 125 | 4番目 |
| 4 | eggs | 70 | 5番目（125のあと） |
学生が入力するセルは、このプレビューでは空白です。
パーティション1つのトピックを作る
curl -s -X POST -H "Content-Type: application/json" -d '{"name":"sales","partitions":1,"replicationFactor":1,"configs":{}}' http://kafka-ui:8080/api/clusters/lab/topics -o /dev/null -w "%{http_code}\n"ここでは表示のみで、実行されません。無料アカウントを作成すると、実行ボタンで自分のラボのターミナルに入力できます。
5件の売上を届いた順に送る
for e in "5 bread" "62 milk" "58 coffee" "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/sales/messages -w "%{http_code}\n"; doneここでは表示のみで、実行されません。無料アカウントを作成すると、実行ボタンで自分のラボのターミナルに入力できます。
オフセットを見る
curl -s http://kafka-ui:8080/api/clusters/lab/topics/sales | grep -o '"partition":[0-9]*\|"offsetMax":[0-9]*'ここでは表示のみで、実行されません。無料アカウントを作成すると、実行ボタンで自分のラボのターミナルに入力できます。
トピックを最初から読む
curl -s -N "http://kafka-ui:8080/api/clusters/lab/topics/sales/messages/v2?seekType=BEGINNING&limit=10" | grep -o '"offset":[0-9]*\|"value":"[^"]*"'ここでは表示のみで、実行されません。無料アカウントを作成すると、実行ボタンで自分のラボのターミナルに入力できます。
ブローカーの時刻印の、時計の分を数える
curl -s -N "http://kafka-ui:8080/api/clusters/lab/topics/sales/messages/v2?seekType=BEGINNING&limit=10" | grep -o '"timestamp":"[^"]*"' | cut -c25-29 | sort -u | wc -lここでは表示のみで、実行されません。無料アカウントを作成すると、実行ボタンで自分のラボのターミナルに入力できます。
窓はイベント時刻の区切り
1分ごとに数えるには、時間を60秒の窓に区切ります。固定の窓は、前の窓が終わった所から始まります。だから、どのイベントもちょうど1つの窓に入ります。
イベントの窓の開始時刻は、時刻から、サイズで割った余りを引いた値です。時刻62なら 62 引く 2 で、窓は60から始まります。
例: 60秒の窓
| 窓 | 開始 | 終わり（含まない） | 商品 |
|---|---|---|---|
| 1つ目 | 0 | 60 | bread, coffee |
| 2つ目 | 60 | 120 | milk, eggs |
| 3つ目 | 120 | 180 | rice |
学生が入力するセルは、このプレビューでは空白です。
窓ごとにイベントを数える
# 各イベントは [イベント時刻の秒, 商品] で、届いた順に並んでいます。
events = [[5, "bread"], [62, "milk"], [58, "coffee"], [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)
# size を 30 に変えて、もう一度実行してみましょう。
トピックそのものから窓を数える
curl -s -N "http://kafka-ui:8080/api/clusters/lab/topics/sales/messages/v2?seekType=BEGINNING&limit=10" | grep -o '"value":"[^"]*"' | cut -d'"' -f4 | while read t item; do echo $((t / 60)); done | sort -n | uniq -cここでは表示のみで、実行されません。無料アカウントを作成すると、実行ボタンで自分のラボのターミナルに入力できます。
よくある間違い
注意
トピックの順番は、出来事が起きた順だと思いがちです。実は届いた順です。窓を選ぶには、イベントの中の時刻を使います。オフセットは、どこまで読んだかを言うためだけに使います。
より新しいイベントのあとに届くと遅れ
コーヒーは58秒に売れましたが、62秒のミルクのあとに届きました。これが遅れて届いたイベントです。より新しいイベントが先に着いたからです。
本物のシステムは、いつまでも待てません。見えた最新の時刻が窓の終わりを過ぎたら、窓を閉じます。少しの許容遅延を持たせると、窓をもう少し長く開けておけます。
遅れて届いたイベントを見つける
# 自分より新しい時刻のイベントが先に届いていたら、遅れて届いたイベントです。
events = [[5, "bread"], [62, "milk"], [58, "coffee"], [125, "rice"], [70, "eggs"]]
newest = 0
for event in events:
if event[0] < newest:
print(event[1], "は遅れて届きました。時刻", event[0], "は", newest, "より前です")
else:
newest = event[0]
# events の最後に [30, "tea"] を足して、もう一度実行してみましょう。
許容遅延のあとで窓を閉じる
# 見えた最新の時刻が、窓の終わりに許容遅延を足した値に届くと、窓は閉じます。
events = [[5, "bread"], [62, "milk"], [58, "coffee"], [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)
# lateness を 0 に、次に 10 に変えて、もう一度実行してみましょう。
ひどく遅れた売上をもう1件送る
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/sales/messages -w "%{http_code}\n"ここでは表示のみで、実行されません。無料アカウントを作成すると、実行ボタンで自分のラボのターミナルに入力できます。
窓をもう一度数える
curl -s -N "http://kafka-ui:8080/api/clusters/lab/topics/sales/messages/v2?seekType=BEGINNING&limit=10" | grep -o '"value":"[^"]*"' | cut -d'"' -f4 | while read t item; do echo $((t / 60)); done | sort -n | uniq -cここでは表示のみで、実行されません。無料アカウントを作成すると、実行ボタンで自分のラボのターミナルに入力できます。
さらに先へ: 許容遅延を選ぶ
許容遅延のコードを、0、5、10で実行します。それぞれ何件捨てるか、書き留めましょう。
そのあと、1文で答えます。店長に出す1日の終わりのレジ報告と、数秒で鳴らす警報があります。遅れて届くイベントを待てるのはどちらですか。それはなぜですか。