| """ |
| Stratified sampling: chα»n 1M users ΔαΊ‘i diα»n tα»« filtered_reviews.jsonl |
| Chiα»u stratify: interaction_count_bucket Γ year_of_last_review |
| |
| Output: |
| sampled_1m_reviews.jsonl β reviews của 1M users ΔΓ£ chα»n |
| sampled_1m_users.json β danh sΓ‘ch user_id + stats |
| sampling_report.json β phΓ’n bα» cells trΖ°α»c/sau sampling |
| """ |
| import json, time, random |
| from collections import defaultdict |
| from pathlib import Path |
|
|
| FILTERED = Path("/workspace/amazon/filtered_reviews.jsonl") |
| OUT_REVIEWS = Path("/workspace/amazon/sampled_1m_reviews.jsonl") |
| OUT_USERS = Path("/workspace/amazon/sampled_1m_users.json") |
| OUT_REPORT = Path("/workspace/amazon/sampling_report.json") |
|
|
| TARGET_USERS = 1_000_000 |
| SEED = 42 |
| random.seed(SEED) |
|
|
| |
| def interaction_bucket(n): |
| if n < 10: return "05-09" |
| if n < 20: return "10-19" |
| if n < 50: return "20-49" |
| if n < 100: return "50-99" |
| return "100+" |
|
|
| def year_bucket(ts): |
| if ts is None: return "unknown" |
| import datetime |
| y = datetime.datetime.fromtimestamp(ts).year |
| if y <= 2012: return "β€2012" |
| if y <= 2014: return "2013-14" |
| if y <= 2016: return "2015-16" |
| return "2017-18" |
|
|
| |
| print("Pass 1: building user stats...", flush=True) |
| t0 = time.time() |
|
|
| user_stats = {} |
|
|
| with open(FILTERED, "r", encoding="utf-8", errors="replace") as f: |
| for i, line in enumerate(f): |
| if i % 10_000_000 == 0 and i > 0: |
| print(f" {i/1e6:.0f}M rows ({time.time()-t0:.0f}s)", flush=True) |
| line = line.strip() |
| if not line: |
| continue |
| try: |
| rec = json.loads(line) |
| except: |
| continue |
| uid = rec.get("reviewerID", "") |
| asin = rec.get("asin", "") |
| ts = rec.get("unixReviewTime") |
| if not uid: |
| continue |
| if uid not in user_stats: |
| user_stats[uid] = {"count": 0, "last_ts": 0, "n_asins": set()} |
| s = user_stats[uid] |
| s["count"] += 1 |
| if ts and ts > s["last_ts"]: |
| s["last_ts"] = ts |
| if asin: |
| s["n_asins"].add(asin) |
|
|
| print(f"Pass 1 done in {time.time()-t0:.1f}s β {len(user_stats):,} users", flush=True) |
|
|
| |
| for uid, s in user_stats.items(): |
| s["n_asins"] = len(s["n_asins"]) |
|
|
| |
| cells = defaultdict(list) |
|
|
| for uid, s in user_stats.items(): |
| ib = interaction_bucket(s["count"]) |
| yb = year_bucket(s["last_ts"] if s["last_ts"] > 0 else None) |
| cells[f"{ib}|{yb}"].append(uid) |
|
|
| print(f"\nCell distribution (before sampling):") |
| total_users = len(user_stats) |
| for cell in sorted(cells): |
| n = len(cells[cell]) |
| print(f" {cell:<20} {n:>8,} ({n/total_users*100:.1f}%)") |
|
|
| |
| print(f"\nProportional sampling β {TARGET_USERS:,} users...", flush=True) |
|
|
| sampled_uids = set() |
| report_cells = {} |
|
|
| for cell, uids in cells.items(): |
| n_cell = len(uids) |
| n_sample = round(n_cell / total_users * TARGET_USERS) |
| n_sample = min(n_sample, n_cell) |
| chosen = random.sample(uids, n_sample) if n_sample < n_cell else uids |
| sampled_uids.update(chosen) |
| report_cells[cell] = {"total": n_cell, "sampled": n_sample} |
|
|
| |
| all_uids = list(user_stats.keys()) |
| if len(sampled_uids) < TARGET_USERS: |
| remaining = [u for u in all_uids if u not in sampled_uids] |
| random.shuffle(remaining) |
| sampled_uids.update(remaining[:TARGET_USERS - len(sampled_uids)]) |
| elif len(sampled_uids) > TARGET_USERS: |
| sampled_list = list(sampled_uids) |
| random.shuffle(sampled_list) |
| sampled_uids = set(sampled_list[:TARGET_USERS]) |
|
|
| print(f"Sampled {len(sampled_uids):,} users", flush=True) |
|
|
| |
| user_records = [ |
| {"uid": uid, **{k: v for k, v in user_stats[uid].items()}} |
| for uid in sampled_uids |
| ] |
| with open(OUT_USERS, "w") as f: |
| json.dump(user_records, f) |
| print(f"Saved {OUT_USERS}", flush=True) |
|
|
| |
| print(f"\nPass 2: writing reviews for sampled users...", flush=True) |
| t1 = time.time() |
| written = 0 |
|
|
| with open(FILTERED, "r", encoding="utf-8", errors="replace") as fin, \ |
| open(OUT_REVIEWS, "w", encoding="utf-8") as fout: |
| for i, line in enumerate(fin): |
| if i % 10_000_000 == 0 and i > 0: |
| print(f" {i/1e6:.0f}M scanned, {written:,} written ({time.time()-t1:.0f}s)", flush=True) |
| line = line.strip() |
| if not line: |
| continue |
| try: |
| rec = json.loads(line) |
| except: |
| continue |
| if rec.get("reviewerID", "") in sampled_uids: |
| fout.write(json.dumps(rec, ensure_ascii=False) + "\n") |
| written += 1 |
|
|
| print(f"Pass 2 done in {time.time()-t1:.1f}s β {written:,} reviews written", flush=True) |
|
|
| |
| sampled_stats = [user_stats[u] for u in sampled_uids] |
| counts = sorted(s["count"] for s in sampled_stats) |
| n = len(counts) |
|
|
| report = { |
| "target_users": TARGET_USERS, |
| "sampled_users": len(sampled_uids), |
| "reviews_written": written, |
| "seed": SEED, |
| "cells": report_cells, |
| "interaction_dist": { |
| "min": counts[0], "p25": counts[n//4], "p50": counts[n//2], |
| "p75": counts[3*n//4], "p90": counts[int(n*.9)], |
| "p99": counts[int(n*.99)], "max": counts[-1], |
| } |
| } |
| with open(OUT_REPORT, "w") as f: |
| json.dump(report, f, indent=2) |
|
|
| print(f"\n{'='*50}") |
| print(f"Done in {time.time()-t0:.1f}s total") |
| print(f" Sampled users : {len(sampled_uids):,}") |
| print(f" Reviews : {written:,}") |
| print(f" interaction p50={counts[n//2]} p90={counts[int(n*.9)]} max={counts[-1]}") |
| print(f" Output : {OUT_REVIEWS}") |
|
|
