S046 · Batch vs Streaming — Mental Model & Use Cases

# batch_vs_stream.py — batch version. Reads a day of events, counts per user per minute.
# Run: python batch_vs_stream.py events_2024_08_10.jsonl
import json, sys, collections, datetime as dt
 
def minute_bucket(ts_iso: str) -> str:
    t = dt.datetime.fromisoformat(ts_iso)
    return t.replace(second=0, microsecond=0).isoformat()
 
def run(path: str) -> None:
    counts: dict[tuple[str,str], int] = collections.Counter()
    with open(path) as f:
        for line in f:
            e = json.loads(line)
            counts[(e["user_id"], minute_bucket(e["ts"]))] += 1
    # Deterministic order for reproducible tests
    for (uid, minute), n in sorted(counts.items()):
        print(f"{minute}\t{uid}\t{n}")
 
if __name__ == "__main__":
    run(sys.argv[1])

# stream_demo.py — streaming version using a socket source.
# Same logic, but incremental. Terminate with Ctrl-C.
# Run:   python stream_demo.py
# Feed:  while true; do echo "{\"user_id\":\"u1\",\"ts\":\"$(date -Iseconds)\"}" | nc -q 0 localhost 9999; sleep 0.5; done
import json, socket, collections, datetime as dt, sys, time
 
WINDOW_SEC = 60
ALLOWED_LATENESS_SEC = 30
 
def minute_bucket(ts: dt.datetime) -> dt.datetime:
    return ts.replace(second=0, microsecond=0)
 
def run(host="0.0.0.0", port=9999) -> None:
    counts: dict[tuple[str, dt.datetime], int] = collections.Counter()
    watermark = dt.datetime.min.replace(tzinfo=dt.timezone.utc)
 
    srv = socket.socket()
    srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    srv.bind((host, port))
    srv.listen(1)
    print(f"listening on {host}:{port} …", file=sys.stderr)
 
    while True:
        conn, _ = srv.accept()
        buf = b""
        while True:
            chunk = conn.recv(4096)
            if not chunk: break
            buf += chunk
            while b"\n" in buf:
                line, buf = buf.split(b"\n", 1)
                if not line.strip(): continue
                e = json.loads(line)
                ts = dt.datetime.fromisoformat(e["ts"])
                key = (e["user_id"], minute_bucket(ts))
 
                # Late data check
                if ts + dt.timedelta(seconds=ALLOWED_LATENESS_SEC) < watermark:
                    print(f"DROP late event ts={ts.isoformat()} watermark={watermark.isoformat()}", file=sys.stderr)
                    continue
 
                counts[key] += 1
                # Advance watermark = max ts seen so far
                if ts > watermark:
                    watermark = ts
 
                # Emit any window that watermark has passed by more than ALLOWED_LATENESS
                cutoff = watermark - dt.timedelta(seconds=ALLOWED_LATENESS_SEC)
                to_emit = [k for k in counts if k[1] + dt.timedelta(seconds=WINDOW_SEC) < cutoff]
                for k in to_emit:
                    print(f"EMIT window={k[1].isoformat()} user={k[0]} n={counts.pop(k)}", flush=True)
        conn.close()
 
if __name__ == "__main__":
    run()