Approach (brief):- Maintain per-1-minute tumbling windows keyed by window_end timestamp.- For each arriving record with event_ts and ingestion_ts compute whether it is late for its event window: if ingestion_ts > window_end + N it’s "too-late" (beyond watermark) — countable as late for percent-late but may be handled differently; if ingestion_ts <= window_end + N it's on-time (arrived before watermark).- Advance watermark = max_seen_event_time - N. When watermark >= window_end, the window is ready to emit: compute percent-late = late_count / total_count, emit alert if > threshold, then evict window state.Pseudocode (Python-style):python
from collections import defaultdict
import math
import time
N = 30 # watermark tolerance in seconds
THRESHOLD = 0.1 # percent-late threshold (10%)
# state
windows_total = defaultdict(int) # key: window_end -> count
windows_late = defaultdict(int)
max_event_ts = 0 # track max event time seen (seconds since epoch)
def window_end_for(ts):
# 1-minute tumbling window aligned on minute
return (ts // 60) * 60 + 60 # exclusive end
def on_event(event_ts, ingestion_ts):
global max_event_ts
max_event_ts = max(max_event_ts, event_ts)
w_end = window_end_for(event_ts)
windows_total[w_end] += 1
# if arrival happened after allowed lateness for this window, it's late
allowed_arrival = w_end + N
if ingestion_ts > allowed_arrival:
windows_late[w_end] += 1
# optionally, also update watermark and trigger emission
advance_and_emit()
def current_watermark():
return max_event_ts - N
def advance_and_emit():
wm = current_watermark()
# find windows whose end <= watermark -> safe to close
ready = [we for we in windows_total.keys() if we <= wm]
for we in ready:
total = windows_total.pop(we, 0)
late = windows_late.pop(we, 0)
pct_late = (late / total) if total > 0 else 0.0
emit_window_result(we, total, late, pct_late)
if pct_late > THRESHOLD:
emit_alert(we, pct_late)
def emit_window_result(window_end, total, late, pct):
print(f"Window ending {window_end}: total={total}, late={late}, pct_late={pct:.2%}")
def emit_alert(window_end, pct):
print(f"ALERT: window {window_end} percent late {pct:.2%} > {THRESHOLD:.2%}")
