Performance Engineering and Cost Optimization Questions

import orjson

# serialize
b = orjson.dumps(obj)           # returns bytes (no extra str copy)
# deserialize
obj = orjson.loads(b)

# streaming encode to a file-like object to avoid building a huge string
import json
def stream_write(obj, fp):
    json.dump(obj, fp)  # writes incrementally to fp

# ijson for streaming parsing of very large JSON
import ijson
with open('big.json', 'rb') as f:
    for item in ijson.items(f, 'records.item'):
        process(item)

# parameters
NUM_BUCKETS = 6            # sliding window buckets (e.g., last 6 minutes)
M = 2_000_000              # bit-array size
K = 4                      # number of hash functions
COUNTER_BITS = 4           # counters per cell (max 15)

# data structures
buckets = [array_of_uint(COUNTER_BITS, M) for _ in range(NUM_BUCKETS)]
current = 0

def hash_indices(id):
    # produce K indices using two independent hashes (double hashing)
    h1 = hash1(id)
    h2 = hash2(id)
    return [(h1 + i*h2) % M for i in range(K)]

def is_duplicate(event_id):
    indices = hash_indices(event_id)
    # present if any bucket has all counters > 0 at those indices
    for b in buckets:
        if all(b[idx] > 0 for idx in indices):
            return True
    return False

def insert(event_id):
    indices = hash_indices(event_id)
    for idx in indices:
        # saturating increment to avoid overflow
        buckets[current][idx] = min(buckets[current][idx] + 1, (1 << COUNTER_BITS) - 1)

def advance_bucket():
    # expire oldest by zeroing the next bucket and rotate
    current = (current + 1) % NUM_BUCKETS
    buckets[current].zero()