To detect hot keys in a high-rate stream within a sliding window, use a time-bucketed counting approach with approximate counters (Count-Min Sketch) per key to save memory and a fixed number of recent buckets to implement a sliding window. Below is a simple in-process prototype using exact bucketed counts (easy to replace with sketches). After code I explain scaling and false-positive reduction.python
import time
from collections import defaultdict, deque
class HotKeyDetector:
def __init__(self, window_secs=60, bucket_secs=1, threshold=1000):
self.window_secs = window_secs
self.bucket_secs = bucket_secs
self.buckets = int(window_secs // bucket_secs)
self.threshold = threshold
# deque of dicts, each dict: key -> count for that bucket
self.ring = deque([defaultdict(int) for _ in range(self.buckets)], maxlen=self.buckets)
self.bucket_start = int(time.time() // bucket_secs)
def _rotate(self):
now_bucket = int(time.time() // self.bucket_secs)
while self.bucket_start < now_bucket:
self.ring.append(defaultdict(int)) # drop oldest, add new empty bucket
self.bucket_start += 1
def ingest(self, key, timestamp=None):
if timestamp is None:
timestamp = time.time()
# assume timestamps non-decreasing or near-real-time; else compute appropriate bucket index
self._rotate()
idx = -1 # newest bucket
self.ring[idx][key] += 1
def is_hot(self, key):
# sum counts across buckets
total = 0
for b in self.ring:
total += b.get(key, 0)
if total > self.threshold:
return True
return False
def top_hot_keys(self):
agg = defaultdict(int)
for b in self.ring:
for k,v in b.items():
agg[k] += v
return [k for k,v in agg.items() if v > self.threshold]
