File size: 3,636 Bytes
e6ea0f1 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 | """
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]}")
|