""" EDA script for Amazon Review + Meta datasets. Streams both files without full extraction to disk. """ import gzip, io, json, sys from collections import Counter, defaultdict from pathlib import Path REVIEW_PATH = Path("/workspace/amazon/All_Amazon_Review_5.json.gz") META_PATH = Path("/workspace/amazon/All_Amazon_Meta.json.gz") SAMPLE_REVIEWS = 500_000 # rows to sample for review EDA SAMPLE_META = 200_000 # rows to sample for meta EDA # ─── helpers ──────────────────────────────────────────────────────────────── def open_review(path): """Single gzip — standard JSONL.""" return gzip.open(path, "rt", encoding="utf-8", errors="replace") def open_meta(path): """Double gzip — outer gz → inner gz → JSONL, fully streaming.""" f_outer = gzip.open(path, "rb") f_inner = gzip.open(f_outer, "rt", encoding="utf-8", errors="replace") return f_inner def stream_jsonl(fh, max_rows, label=""): for i, line in enumerate(fh): if i >= max_rows: break if i % 100_000 == 0 and i > 0: print(f" [{label}] {i:,} rows...", flush=True) line = line.strip() if not line: continue try: yield json.loads(line) except json.JSONDecodeError: continue # ─── Review EDA ───────────────────────────────────────────────────────────── def eda_reviews(): print("\n" + "="*60) print("REVIEW FILE EDA") print("="*60) fields_seen = Counter() overall_dist = Counter() verified_dist = Counter() user_counts = Counter() item_counts = Counter() year_counts = Counter() has_text = 0 has_vote = 0 total = 0 with open_review(REVIEW_PATH) as fh: for rec in stream_jsonl(fh, SAMPLE_REVIEWS, "review"): total += 1 for k in rec: fields_seen[k] += 1 overall_dist[int(rec.get("overall", 0))] += 1 verified_dist[rec.get("verified", None)] += 1 user_counts[rec.get("reviewerID", "")] += 1 item_counts[rec.get("asin", "")] += 1 ts = rec.get("unixReviewTime") if ts: import datetime year_counts[datetime.datetime.fromtimestamp(ts).year] += 1 if rec.get("reviewText", "").strip(): has_text += 1 if rec.get("vote"): has_vote += 1 print(f"\nRows sampled : {total:,}") print(f"Unique users : {len(user_counts):,}") print(f"Unique items : {len(item_counts):,}") print("\nField presence (%):") for f, cnt in sorted(fields_seen.items(), key=lambda x: -x[1]): print(f" {f:<20} {cnt/total*100:6.1f}%") print("\nRating distribution:") for star in sorted(overall_dist): bar = "#" * int(overall_dist[star] / total * 50) print(f" {star}★ {overall_dist[star]:>7,} {bar}") print(f"\nVerified purchases: {verified_dist.get(True,0)/total*100:.1f}%") print(f"Has review text : {has_text/total*100:.1f}%") print(f"Has vote field : {has_vote/total*100:.1f}%") print("\nYear distribution (sampled):") for yr in sorted(year_counts): bar = "#" * int(year_counts[yr] / total * 40) print(f" {yr} {year_counts[yr]:>7,} {bar}") print("\nUser review count distribution:") cnt_vals = list(user_counts.values()) cnt_vals.sort() n = len(cnt_vals) print(f" min={cnt_vals[0]} p25={cnt_vals[n//4]} median={cnt_vals[n//2]} " f"p75={cnt_vals[3*n//4]} p90={cnt_vals[int(n*.9)]} " f"p99={cnt_vals[int(n*.99)]} max={cnt_vals[-1]}") print(f" Users with >=5 reviews: {sum(1 for v in cnt_vals if v>=5):,} " f"({sum(1 for v in cnt_vals if v>=5)/n*100:.1f}%)") print("\nTop-10 most reviewed items:") for asin, cnt in item_counts.most_common(10): print(f" {asin} {cnt:,}") # ─── Meta EDA ─────────────────────────────────────────────────────────────── def eda_meta(): print("\n" + "="*60) print("META FILE EDA (double-gzip — loading outer layer first...)") print("="*60) fields_seen = Counter() has_price = 0 has_also_buy = 0 has_also_view = 0 has_desc = 0 has_image = 0 cat_top = Counter() price_vals = [] also_buy_lens = [] also_view_lens = [] total = 0 with open_meta(META_PATH) as fh: for rec in stream_jsonl(fh, SAMPLE_META, "meta"): total += 1 for k in rec: fields_seen[k] += 1 p = rec.get("price") if p not in (None, "", "None"): has_price += 1 try: price_vals.append(float(str(p).replace("$","").replace(",",""))) except ValueError: pass ab = rec.get("also_buy") or [] av = rec.get("also_view") or [] desc = rec.get("description") or [] if ab: has_also_buy += 1 also_buy_lens.append(len(ab)) if av: has_also_view += 1 also_view_lens.append(len(av)) if desc and any(d.strip() for d in (desc if isinstance(desc, list) else [desc])): has_desc += 1 if rec.get("imageURL") or rec.get("imageURLHighRes"): has_image += 1 cats = rec.get("category") or [] if cats: cat_top[cats[0]] += 1 print(f"\nRows sampled : {total:,}") print("\nField presence (%):") for f, cnt in sorted(fields_seen.items(), key=lambda x: -x[1]): print(f" {f:<20} {cnt/total*100:6.1f}%") print(f"\nHas price : {has_price/total*100:.1f}%") print(f"Has also_buy : {has_also_buy/total*100:.1f}%") print(f"Has also_view : {has_also_view/total*100:.1f}%") print(f"Has description : {has_desc/total*100:.1f}%") print(f"Has image : {has_image/total*100:.1f}%") if price_vals: price_vals.sort() n = len(price_vals) print(f"\nPrice stats (USD):") print(f" min={price_vals[0]:.2f} p25={price_vals[n//4]:.2f} " f"median={price_vals[n//2]:.2f} p75={price_vals[3*n//4]:.2f} " f"p90={price_vals[int(n*.9)]:.2f} p99={price_vals[int(n*.99)]:.2f} " f"max={price_vals[-1]:.2f}") if also_buy_lens: also_buy_lens.sort() n = len(also_buy_lens) print(f"\nalso_buy length: mean={sum(also_buy_lens)/n:.1f} " f"median={also_buy_lens[n//2]} max={also_buy_lens[-1]}") if also_view_lens: also_view_lens.sort() n = len(also_view_lens) print(f"also_view length: mean={sum(also_view_lens)/n:.1f} " f"median={also_view_lens[n//2]} max={also_view_lens[-1]}") print("\nTop-15 categories (first-level):") for cat, cnt in cat_top.most_common(15): bar = "#" * int(cnt / total * 40) print(f" {cat[:50]:<52} {cnt:>7,} {bar}") print("\nSample records:") with open_meta(META_PATH) as fh: for i, rec in enumerate(stream_jsonl(fh, 3, "meta-sample")): print(f"\n--- Record {i+1} ---") for k, v in rec.items(): val_str = str(v)[:120] print(f" {k:<20}: {val_str}") # ─── main ──────────────────────────────────────────────────────────────────── if __name__ == "__main__": mode = sys.argv[1] if len(sys.argv) > 1 else "all" if mode in ("all", "review"): eda_reviews() if mode in ("all", "meta"): eda_meta() print("\nDone.")