""" Count users with verified=True and >=5 interactions. Streams All_Amazon_Review_5.json.gz without full extraction. """ import gzip, json, time from collections import Counter REVIEW_PATH = "/workspace/amazon/All_Amazon_Review_5.json.gz" MIN_INTERACTIONS = 5 print("Streaming review file...", flush=True) t0 = time.time() user_counts = Counter() # user -> total verified interactions user_items = {} # user -> set of unique ASINs (verified only) total_rows = 0 skipped_unverified = 0 with gzip.open(REVIEW_PATH, "rt", encoding="utf-8", errors="replace") as f: for line in f: line = line.strip() if not line: continue try: rec = json.loads(line) except json.JSONDecodeError: continue total_rows += 1 if total_rows % 5_000_000 == 0: elapsed = time.time() - t0 eligible = sum(1 for v in user_counts.values() if v >= MIN_INTERACTIONS) print(f" {total_rows/1e6:.0f}M rows | {elapsed:.0f}s | " f"eligible users so far: {eligible:,}", flush=True) if not rec.get("verified", False): skipped_unverified += 1 continue uid = rec.get("reviewerID", "") asin = rec.get("asin", "") if not uid or not asin: continue user_counts[uid] += 1 if uid not in user_items: user_items[uid] = set() user_items[uid].add(asin) elapsed = time.time() - t0 print(f"\nDone in {elapsed:.1f}s", flush=True) # ── Stats ────────────────────────────────────────────────────────────────── total_users = len(user_counts) eligible_users = {u for u, c in user_counts.items() if c >= MIN_INTERACTIONS} n_eligible = len(eligible_users) eligible_reviews = sum(user_counts[u] for u in eligible_users) print(f"\n{'='*55}") print(f"RESULTS") print(f"{'='*55}") print(f"Total rows scanned : {total_rows:>12,}") print(f"Skipped (unverified) : {skipped_unverified:>12,} ({skipped_unverified/total_rows*100:.1f}%)") print(f"Verified rows : {total_rows-skipped_unverified:>12,} ({(total_rows-skipped_unverified)/total_rows*100:.1f}%)") print(f"\nUnique users (verified) : {total_users:>12,}") print(f"Users with >=5 verified : {n_eligible:>12,} ({n_eligible/total_users*100:.1f}% of verified users)") print(f"Reviews from eligible : {eligible_reviews:>12,}") # interaction count distribution among eligible users counts = sorted(user_counts[u] for u in eligible_users) n = len(counts) if n: percentiles = [25, 50, 75, 90, 95, 99] print(f"\nInteraction distribution (eligible users only):") print(f" min={counts[0]} max={counts[-1]}") for p in percentiles: idx = min(int(n * p / 100), n - 1) print(f" p{p:>2} = {counts[idx]}") # bucket distribution print(f"\nBucket breakdown (eligible users):") buckets = [(5,9),(10,19),(20,49),(50,99),(100,499),(500,9999)] for lo, hi in buckets: cnt = sum(1 for v in counts if lo <= v <= hi) print(f" {lo:>4}–{hi:<4} interactions: {cnt:>8,} ({cnt/n_eligible*100:.1f}%)") # unique items per eligible user unique_item_counts = sorted(len(user_items[u]) for u in eligible_users) print(f"\nUnique items/user distribution (eligible):") print(f" min={unique_item_counts[0]} " f"p50={unique_item_counts[n//2]} " f"p90={unique_item_counts[int(n*.9)]} " f"p99={unique_item_counts[int(n*.99)]} " f"max={unique_item_counts[-1]}")