Approach:Use the space-saving (aka weighted "heavy hitters") algorithm to stream NetFlow records and maintain an approximate top-k by total bytes using O(k) space. This avoids storing all IPs (required since file doesn't fit in memory). The algorithm keeps up to k counters (src_ip → estimated_bytes). For each record with weight w (bytes):- If src_ip in counters: increment its counter by w.- Else if fewer than k counters: add src_ip with count w and error=0.- Else: replace the entry with the current minimum counter: set its src_ip to current src_ip, its count = min_count + w, and its error = min_count (so we track possible overestimate).This yields good practical accuracy for top talkers and runs in one pass.Code implementation:python
import csv
import heapq
def top_k_src_ips_by_bytes_stream(csv_path, k=10, bytes_field='bytes'):
"""
Space-saving algorithm for top-k src_ip by total bytes in a CSV stream.
Returns list of (src_ip, estimated_bytes) sorted descending.
"""
# counters: dict src_ip -> [count, error]
counters = {}
with open(csv_path, newline='') as f:
reader = csv.DictReader(f)
for row in reader:
try:
src = row['src_ip']
w = int(row.get(bytes_field, 0))
except (KeyError, ValueError):
continue # skip malformed rows
if src in counters:
counters[src][0] += w
elif len(counters) < k:
counters[src] = [w, 0]
else:
# find key with minimum count
min_key = min(counters, key=lambda x: counters[x][0])
min_count, min_err = counters[min_key]
# replace it with current src
del counters[min_key]
# new count is min_count + weight; record error = min_count
counters[src] = [min_count + w, min_count]
# return sorted list by estimated bytes (count), descending
result = sorted(((ip, c_e[0]) for ip, c_e in counters.items()),
key=lambda x: x[1], reverse=True)
return result[:k]
