diff --git a/benchmarks/gfql/bench_graphframes.py b/benchmarks/gfql/bench_graphframes.py new file mode 100644 index 0000000000..e2337ab36b --- /dev/null +++ b/benchmarks/gfql/bench_graphframes.py @@ -0,0 +1,691 @@ +#!/usr/bin/env python3 +""" +bench_graphframes.py — GFQL vs Apache Spark GraphFrames benchmark harness. + +Compares GFQL (graphistry's dataframe-native graph query language) against +Spark GraphFrames on large SNAP graphs across three tasks: attribute/degree +filter, 1- and 2-hop neighborhood traversal, and full-graph PageRank. + +DESIGN GOALS +------------ +- Single file, no deps beyond graphistry / pyspark / graphframes / pandas / polars. +- Every (system, task) is *guarded*: an error/OOM in one cell records + {"status": "error", ...} and the run continues. Missing GraphFrames / + pyspark / GPU is skipped with a clear message, never aborts. +- Timing: `--warmups` warmup iterations (default 2) then `--iters` timed + iterations (default 5); we report the *median* wall-clock ms. Cold load + (parquet/txt -> in-memory graph) is timed once per system. +- Results stream to JSONL, one line per (system, task, dataset). + +This script is meant to be *reviewed and then run on the benchmark box* +(datasets live at ~/data/snap). It does not download or install anything. + +Task definitions and fairness caveats are documented in DESIGN.md. +""" + +from __future__ import annotations + +import argparse +import gc +import json +import os +import statistics +import sys +import time +import traceback +from dataclasses import dataclass, field, asdict +from typing import Any, Callable, Dict, List, Optional, Tuple + + +# --------------------------------------------------------------------------- +# Dataset registry +# --------------------------------------------------------------------------- +# Each dataset has a parquet file (fast path) and a gzipped SNAP txt fallback. +# SNAP edge-list format: tab-separated "srcdst", comment lines start '#', +# undirected. Parquet column names are configurable via --src-col / --dst-col +# because SNAP-derived parquet files in the wild use either ('src','dst') or +# the raw SNAP header names; we default to ('src','dst') and auto-detect below. +DATASETS: Dict[str, Dict[str, str]] = { + "lj": { + "parquet": "com-lj.ungraph.txt.gz.parquet", + "txt": "com-lj.ungraph.txt.gz", + "approx_edges": "35M", + }, + "orkut": { + "parquet": "com-orkut.ungraph.txt.gz.parquet", + "txt": "com-orkut.ungraph.txt.gz", + "approx_edges": "117M", + }, + "friendster": { + # No prebuilt parquet shipped for friendster; txt (~1.8B edges) only. + "parquet": "com-friendster.ungraph.txt.gz.parquet", + "txt": "com-friendster.ungraph.txt.gz", + "approx_edges": "1.8B", + }, +} + +ALL_SYSTEMS = ["gfql-polars", "gfql-polars-gpu", "graphframes"] +ALL_TASKS = ["filter", "hop1", "hop2", "pagerank"] + + +# --------------------------------------------------------------------------- +# Result record +# --------------------------------------------------------------------------- +@dataclass +class Result: + system: str + task: str + dataset: str + n_edges: Optional[int] = None + n_nodes: Optional[int] = None + median_ms: Optional[float] = None + cold_load_ms: Optional[float] = None + iters: int = 0 + warmups: int = 0 + result_size: Optional[int] = None # rows materialized (for parity checks) + status: str = "ok" + error: Optional[str] = None + extra: Dict[str, Any] = field(default_factory=dict) + + def to_jsonl(self) -> str: + return json.dumps(asdict(self)) + + +# --------------------------------------------------------------------------- +# Timing helpers +# --------------------------------------------------------------------------- +def _time_once(fn: Callable[[], Any]) -> Tuple[float, Any]: + """Run fn(), returning (elapsed_ms, return_value).""" + t0 = time.perf_counter() + out = fn() + t1 = time.perf_counter() + return (t1 - t0) * 1000.0, out + + +def timed_median( + fn: Callable[[], Any], + warmups: int, + iters: int, +) -> Tuple[float, Optional[int]]: + """ + Warm up `warmups` times (untimed), then time `iters` runs. + Returns (median_ms, last_result_size). + + `fn` must return something we can size for a parity check: we accept an int + directly, or fall back to len(). None -> result_size None. + """ + for _ in range(warmups): + fn() + gc.collect() + samples: List[float] = [] + last_size: Optional[int] = None + for _ in range(iters): + ms, out = _time_once(fn) + samples.append(ms) + last_size = _sizeof(out) + gc.collect() + return statistics.median(samples), last_size + + +def _sizeof(out: Any) -> Optional[int]: + if out is None: + return None + if isinstance(out, int): + return out + try: + return len(out) + except TypeError: + return None + + +# --------------------------------------------------------------------------- +# GFQL system adapter +# --------------------------------------------------------------------------- +class GFQLSystem: + """ + GFQL adapter. `engine` is 'polars' or 'polars-gpu' (see graphistry.Engine). + + Cold load: read parquet/txt -> pandas/polars edge frame -> build a + graphistry Plottable with a precomputed node table carrying `degree` + (so the filter task is a pure node-attribute WHERE, symmetric with the + GraphFrames `gf.degrees` filter). The degree precompute is part of cold + load for both systems. + """ + + def __init__(self, engine: str, src_col: str, dst_col: str, + filter_threshold: Optional[int] = None): + self.engine = engine # 'polars' | 'polars-gpu' + self.src_col = src_col + self.dst_col = dst_col + self._threshold_override = filter_threshold + self.g = None + self.n_edges: Optional[int] = None + self.n_nodes: Optional[int] = None + self._seeds: List[Any] = [] + self._filter_threshold: Optional[int] = None + + # -- cold load ---------------------------------------------------------- + def load(self, edges_path: str, is_parquet: bool) -> None: + import pandas as pd + import graphistry + + if is_parquet: + edf = pd.read_parquet(edges_path) + edf = self._normalize_cols(edf) + else: + # SNAP gz txt: tab-separated, comments start with '#'. + edf = pd.read_csv( + edges_path, + sep="\t", + comment="#", + header=None, + names=[self.src_col, self.dst_col], + compression="gzip", + dtype="int64", + ) + + # Node table with degree (undirected: count endpoints on both sides). + deg = ( + pd.concat([edf[self.src_col], edf[self.dst_col]]) + .value_counts() + .rename_axis("id") + .reset_index(name="degree") + ) + self.n_edges = int(len(edf)) + self.n_nodes = int(len(deg)) + + g = graphistry.edges(edf, self.src_col, self.dst_col) + g = g.nodes(deg, "id") + self.g = g + + # Filter threshold: shared override (for cross-system parity) else + # ~top-decile degree. Seeds: highest-degree nodes. + self._filter_threshold = ( + self._threshold_override + if self._threshold_override is not None + else int(deg["degree"].quantile(0.90)) + ) + self._seeds = deg.sort_values("degree", ascending=False)["id"].head(50).tolist() + + def _normalize_cols(self, edf): + """Map the parquet's actual columns onto (src_col, dst_col).""" + cols = list(edf.columns) + if self.src_col in cols and self.dst_col in cols: + return edf + # Fall back to positional: first two columns are src, dst. + if len(cols) >= 2: + return edf.rename(columns={cols[0]: self.src_col, cols[1]: self.dst_col}) + raise ValueError(f"parquet has too few columns: {cols}") + + # -- task fns (each returns a size for parity) -------------------------- + def filter_fn(self) -> Callable[[], int]: + from graphistry.compute.ast import n + from graphistry.compute.predicates.numeric import ge + + g, engine, thr = self.g, self.engine, self._filter_threshold + + def run() -> int: + out = g.gfql([n(filter_dict={"degree": ge(thr)})], engine=engine) + return int(len(out._nodes)) # materialize + + return run + + def hop_fn(self, hops: int) -> Callable[[], int]: + from graphistry.compute.ast import n, e_undirected + from graphistry.compute.predicates.is_in import IsIn + + g, engine, seeds = self.g, self.engine, self._seeds + + def run() -> int: + out = g.gfql( + [ + n({"id": IsIn(options=seeds)}), + e_undirected(to_fixed_point=False, hops=hops), + n(), + ], + engine=engine, + ) + # Return the k-ball NODE count only, so it is directly comparable to + # the GraphFrames `visited.count()` (union of seeds + up-to-k-hop + # neighbors). Edges are still materialized by the traversal; we just + # do not fold them into the parity size. + return int(len(out._nodes)) + + return run + + def pagerank_fn(self) -> Callable[[], int]: + """ + PageRank API (found in repo): + - CPU (polars / pandas): g.compute_igraph('pagerank') + graphistry/plugins/igraph.py:339 + - GPU: g.compute_cugraph('pagerank') + graphistry/plugins/cugraph.py:423 + polars has no native PageRank; the polars engine routes PageRank + through igraph (pandas conversion under the hood). + """ + g = self.g + use_gpu = self.engine == "polars-gpu" + + def run() -> int: + if use_gpu: + out = g.compute_cugraph("pagerank") + else: + out = g.compute_igraph("pagerank") + return int(len(out._nodes)) # materialize the pagerank column + + return run + + +# --------------------------------------------------------------------------- +# GraphFrames system adapter +# --------------------------------------------------------------------------- +class GraphFramesSystem: + """ + Spark GraphFrames adapter. SparkSession is local[*] (multicore single node) + with configurable driver memory. Imports are guarded so a box without + pyspark/graphframes skips gracefully. + """ + + def __init__(self, src_col: str, dst_col: str, spark_mem: str, pagerank_iters: int, + spark_jars: Optional[str] = None, filter_threshold: Optional[int] = None): + self.src_col = src_col + self.dst_col = dst_col + self.spark_mem = spark_mem + self.pagerank_iters = pagerank_iters + self._threshold_override = filter_threshold + # Path(s) to the graphframes assembly jar. Without it the JVM has no + # GraphFrame classes and the first Spark action fails. Falls back to the + # GRAPHFRAMES_JAR env var so the documented run command can stay short. + self.spark_jars = spark_jars or os.environ.get("GRAPHFRAMES_JAR") + self.spark = None + self.gf = None + self.n_edges: Optional[int] = None + self.n_nodes: Optional[int] = None + self._seeds: List[Any] = [] + self._filter_threshold: Optional[int] = None + + # -- cold load ---------------------------------------------------------- + def load(self, edges_path: str, is_parquet: bool) -> None: + from pyspark.sql import SparkSession + from pyspark.sql import functions as F + from graphframes import GraphFrame # guarded by caller + + builder = ( + SparkSession.builder.appName("gfql-vs-graphframes") + .master("local[*]") + .config("spark.driver.memory", self.spark_mem) + .config("spark.sql.shuffle.partitions", str(os.cpu_count() or 8)) + # GraphFrames connected-components / pageRank checkpointing: + .config("spark.sql.adaptive.enabled", "true") + ) + if self.spark_jars: + builder = builder.config("spark.jars", self.spark_jars) + self.spark = builder.getOrCreate() + # Checkpoint dir is required by some GraphFrames algorithms. + self.spark.sparkContext.setCheckpointDir( + os.path.join(os.path.expanduser("~"), ".spark-checkpoints") + ) + + if is_parquet: + edf = self.spark.read.parquet(edges_path) + cols = edf.columns + if not (self.src_col in cols and self.dst_col in cols): + edf = edf.withColumnRenamed(cols[0], self.src_col).withColumnRenamed( + cols[1], self.dst_col + ) + else: + # SNAP gz txt: tab-separated with '#' comments. + edf = ( + self.spark.read.option("sep", "\t") + .option("comment", "#") + .csv(edges_path) + ) + edf = ( + edf.withColumnRenamed("_c0", self.src_col) + .withColumnRenamed("_c1", self.dst_col) + .select( + F.col(self.src_col).cast("long"), + F.col(self.dst_col).cast("long"), + ) + ) + + # GraphFrames wants columns named 'src','dst' on edges and 'id' on nodes. + edges = edf.select( + F.col(self.src_col).alias("src"), F.col(self.dst_col).alias("dst") + ).cache() + vertices = ( + edges.select(F.col("src").alias("id")) + .union(edges.select(F.col("dst").alias("id"))) + .distinct() + .cache() + ) + self.gf = GraphFrame(vertices, edges) + self.n_edges = edges.count() # force materialization -> honest load + self.n_nodes = vertices.count() + + # Degree threshold + seeds computed from gf.degrees (materialized once). + degrees = self.gf.degrees.cache() + # ~top-decile degree via approxQuantile (exact quantile is expensive). + # A shared override (--filter-threshold) is preferred so the filter task + # is bit-identical across systems for a clean parity check. + if self._threshold_override is not None: + self._filter_threshold = self._threshold_override + else: + thr = degrees.approxQuantile("degree", [0.90], 0.01) + self._filter_threshold = int(thr[0]) if thr else 1 + top = ( + degrees.orderBy(F.col("degree").desc()).limit(50).select("id").collect() + ) + self._seeds = [r["id"] for r in top] + self._degrees = degrees + + # -- task fns ----------------------------------------------------------- + def filter_fn(self) -> Callable[[], int]: + thr = self._filter_threshold + degrees = self._degrees + + def run() -> int: + # WHERE on the degree column; .count() forces materialization. + return degrees.filter(degrees["degree"] >= thr).count() + + return run + + def hop_fn(self, hops: int) -> Callable[[], int]: + """ + k-hop neighborhood from seeds. GraphFrames has no direct k-hop + neighborhood op (bfs finds shortest paths between predicates, motif + `find` matches fixed patterns), so we expand via iterated edge joins + against the (undirected) edge frame — still pure Spark, honest timing. + """ + from pyspark.sql import functions as F + + spark = self.spark + edges = self.gf.edges + seeds = self._seeds + + # Undirected adjacency: both directions. + adj = edges.select("src", "dst").union( + edges.select(F.col("dst").alias("src"), F.col("src").alias("dst")) + ).cache() + + def run() -> int: + frontier = spark.createDataFrame([(s,) for s in seeds], ["id"]).distinct() + visited = frontier + for _ in range(hops): + nxt = ( + frontier.join(adj, frontier["id"] == adj["src"]) + .select(F.col("dst").alias("id")) + .distinct() + ) + visited = visited.union(nxt).distinct() + frontier = nxt + return visited.count() # materialize the neighborhood + + return run + + def pagerank_fn(self) -> Callable[[], int]: + gf = self.gf + iters = self.pagerank_iters + + def run() -> int: + # resetProbability = 1 - damping (0.85) = 0.15 + res = gf.pageRank(resetProbability=0.15, maxIter=iters) + return res.vertices.count() # materialize + + return run + + def close(self) -> None: + if self.spark is not None: + try: + self.spark.stop() + except Exception: + pass + + +# --------------------------------------------------------------------------- +# Runner +# --------------------------------------------------------------------------- +def resolve_edges_path(data_dir: str, dataset: str) -> Tuple[str, bool]: + """Return (path, is_parquet). Prefer parquet, fall back to gz txt.""" + d = DATASETS[dataset] + pq = os.path.join(data_dir, d["parquet"]) + txt = os.path.join(data_dir, d["txt"]) + if os.path.exists(pq): + return pq, True + if os.path.exists(txt): + return txt, False + # Return the parquet path anyway; loader will raise a clear FileNotFound. + return pq, True + + +def run_system( + system: str, + tasks: List[str], + dataset: str, + edges_path: str, + is_parquet: bool, + args: argparse.Namespace, + out_fh, +) -> List[Result]: + """Load one system, run each task guarded, stream JSONL results.""" + # --- construct + guarded import/availability --------------------------- + adapter = None + cold_ms: Optional[float] = None + n_edges = n_nodes = None + load_error: Optional[str] = None + rows: List[Result] = [] + + try: + if system in ("gfql-polars", "gfql-polars-gpu"): + engine = "polars" if system == "gfql-polars" else "polars-gpu" + import graphistry # noqa: F401 (fail fast if missing) + adapter = GFQLSystem(engine, args.src_col, args.dst_col, + filter_threshold=args.filter_threshold) + elif system == "graphframes": + # Guarded imports: absence -> skip cleanly. + import pyspark # noqa: F401 + from graphframes import GraphFrame # noqa: F401 + adapter = GraphFramesSystem( + args.src_col, args.dst_col, args.spark_mem, args.pagerank_iters, + spark_jars=args.spark_jars, filter_threshold=args.filter_threshold, + ) + else: + raise ValueError(f"unknown system {system}") + + cold_ms, _ = _time_once(lambda: adapter.load(edges_path, is_parquet)) + n_edges = adapter.n_edges + n_nodes = adapter.n_nodes + except ImportError as e: + load_error = f"skipped (import failed): {e}" + print(f"[SKIP] {system}: {load_error}", file=sys.stderr) + except Exception as e: + load_error = f"cold-load failed: {e}\n{traceback.format_exc()}" + print(f"[ERROR] {system} cold load: {e}", file=sys.stderr) + + # If load failed, emit one error row per requested task and return. + if adapter is None or load_error is not None: + for task in tasks: + r = Result( + system=system, task=task, dataset=dataset, + cold_load_ms=cold_ms, iters=0, warmups=args.warmups, + status="error", error=load_error or "unavailable", + ) + out_fh.write(r.to_jsonl() + "\n") + out_fh.flush() + rows.append(r) + return rows + + # --- run each task, guarded ------------------------------------------- + for task in tasks: + r = Result( + system=system, task=task, dataset=dataset, + n_edges=n_edges, n_nodes=n_nodes, cold_load_ms=cold_ms, + iters=args.iters, warmups=args.warmups, + ) + try: + fn = build_task_fn(adapter, task) + median_ms, size = timed_median(fn, args.warmups, args.iters) + r.median_ms = median_ms + r.result_size = size + r.status = "ok" + print( + f"[OK] {system}/{task}/{dataset}: " + f"median={median_ms:.1f}ms size={size}", + file=sys.stderr, + ) + except Exception as e: + r.status = "error" + r.error = f"{e}\n{traceback.format_exc()}" + print(f"[ERROR] {system}/{task}: {e}", file=sys.stderr) + out_fh.write(r.to_jsonl() + "\n") + out_fh.flush() + rows.append(r) + + # --- teardown ---------------------------------------------------------- + if isinstance(adapter, GraphFramesSystem): + adapter.close() + return rows + + +def build_task_fn(adapter, task: str) -> Callable[[], int]: + if task == "filter": + return adapter.filter_fn() + if task == "hop1": + return adapter.hop_fn(1) + if task == "hop2": + return adapter.hop_fn(2) + if task == "pagerank": + return adapter.pagerank_fn() + raise ValueError(f"unknown task {task}") + + +def validate_result_size_parity(rows: List[Result]) -> List[str]: + """Return parity errors for successful rows grouped by dataset/task. + + Missing systems and error rows are allowed so exploratory sweeps still finish, + but any successful systems for the same task must report identical sizes. + """ + grouped: Dict[Tuple[str, str], Dict[str, Optional[int]]] = {} + for row in rows: + if row.status != "ok": + continue + grouped.setdefault((row.dataset, row.task), {})[row.system] = row.result_size + + errors: List[str] = [] + for (dataset, task), sizes_by_system in sorted(grouped.items()): + if len(sizes_by_system) < 2: + continue + sizes = set(sizes_by_system.values()) + if len(sizes) > 1: + detail = ", ".join( + f"{system}={size}" for system, size in sorted(sizes_by_system.items()) + ) + errors.append(f"{dataset}/{task} result_size mismatch: {detail}") + return errors + + +# --------------------------------------------------------------------------- +# CLI +# --------------------------------------------------------------------------- +def parse_args(argv: Optional[List[str]] = None) -> argparse.Namespace: + p = argparse.ArgumentParser( + description="GFQL vs Spark GraphFrames benchmark harness (SNAP graphs).", + formatter_class=argparse.ArgumentDefaultsHelpFormatter, + ) + p.add_argument("--dataset", required=True, choices=list(DATASETS.keys())) + p.add_argument("--data-dir", default="~/data/snap") + p.add_argument( + "--systems", default=",".join(ALL_SYSTEMS), + help="comma list from: " + ", ".join(ALL_SYSTEMS), + ) + p.add_argument( + "--tasks", default=",".join(ALL_TASKS), + help="comma list from: " + ", ".join(ALL_TASKS), + ) + p.add_argument("--warmups", type=int, default=2) + p.add_argument("--iters", type=int, default=5) + p.add_argument("--out", default="results.jsonl") + p.add_argument("--spark-mem", default="200g", help="Spark driver memory") + p.add_argument( + "--pagerank-iters", type=int, default=20, + help="maxIter for GraphFrames pageRank (fixed-iteration for parity)", + ) + p.add_argument( + "--filter-threshold", type=int, default=None, + help="shared degree>=T for the filter task (identical across systems " + "for a clean parity check); default: each computes its own p90", + ) + p.add_argument( + "--spark-jars", default=None, + help="path to graphframes assembly jar (else GRAPHFRAMES_JAR env)", + ) + p.add_argument("--src-col", default="src", help="edge source column name") + p.add_argument("--dst-col", default="dst", help="edge destination column name") + p.add_argument( + "--dry-run", action="store_true", + help="print the plan and exit without executing", + ) + return p.parse_args(argv) + + +def main(argv: Optional[List[str]] = None) -> int: + args = parse_args(argv) + data_dir = os.path.expanduser(args.data_dir) + systems = [s.strip() for s in args.systems.split(",") if s.strip()] + tasks = [t.strip() for t in args.tasks.split(",") if t.strip()] + + bad_sys = [s for s in systems if s not in ALL_SYSTEMS] + bad_task = [t for t in tasks if t not in ALL_TASKS] + if bad_sys: + print(f"unknown systems: {bad_sys}", file=sys.stderr) + return 2 + if bad_task: + print(f"unknown tasks: {bad_task}", file=sys.stderr) + return 2 + + edges_path, is_parquet = resolve_edges_path(data_dir, args.dataset) + + # -- plan summary ------------------------------------------------------- + print("=" * 70, file=sys.stderr) + print("GFQL vs GraphFrames benchmark plan", file=sys.stderr) + print(f" dataset : {args.dataset} (~{DATASETS[args.dataset]['approx_edges']} edges)", file=sys.stderr) + print(f" edges_path : {edges_path} (parquet={is_parquet})", file=sys.stderr) + print(f" cols : src={args.src_col} dst={args.dst_col}", file=sys.stderr) + print(f" systems : {systems}", file=sys.stderr) + print(f" tasks : {tasks}", file=sys.stderr) + print(f" warmups : {args.warmups} iters(median of): {args.iters}", file=sys.stderr) + print(f" spark_mem : {args.spark_mem} pagerank_iters: {args.pagerank_iters}", file=sys.stderr) + print(f" out : {args.out}", file=sys.stderr) + print("=" * 70, file=sys.stderr) + + if args.dry_run: + print("[DRY RUN] no execution; exiting.", file=sys.stderr) + return 0 + + if not os.path.exists(edges_path): + print( + f"[WARN] edges path not found: {edges_path} — systems will record errors.", + file=sys.stderr, + ) + + all_rows: List[Result] = [] + with open(args.out, "w") as out_fh: + for system in systems: + print(f"\n### SYSTEM: {system} ###", file=sys.stderr) + all_rows.extend( + run_system(system, tasks, args.dataset, edges_path, is_parquet, args, out_fh) + ) + + print(f"\nDone. Results -> {args.out}", file=sys.stderr) + parity_errors = validate_result_size_parity(all_rows) + if parity_errors: + print("[PARITY ERROR] result-size mismatch across successful systems:", file=sys.stderr) + for err in parity_errors: + print(f" - {err}", file=sys.stderr) + return 4 + print("[PARITY OK] result sizes match across successful systems.", file=sys.stderr) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/benchmarks/gfql/bench_graphframes_DESIGN.md b/benchmarks/gfql/bench_graphframes_DESIGN.md new file mode 100644 index 0000000000..71504efbab --- /dev/null +++ b/benchmarks/gfql/bench_graphframes_DESIGN.md @@ -0,0 +1,74 @@ +# GFQL vs Spark GraphFrames — benchmark design + +Harness: `bench_graphframes.py`. Compares GFQL (dataframe-native, single-node +columnar) against Spark GraphFrames (JVM, `local[*]`) on SNAP graphs. + +## Tasks (same semantics on both systems) + +- **filter** — a `WHERE` on a numeric column. We precompute node `degree` at + cold-load and filter nodes with `degree >= p90(degree)`. GFQL: + `g.gfql([n(filter_dict={'degree': ge(thr)})])`. GraphFrames: + `gf.degrees.filter(degree >= thr).count()`. SNAP graphs carry no attributes, + so degree is the natural threshold column; the degree precompute is charged + to cold-load for *both* systems. +- **hop1 / hop2** — 1- and 2-hop undirected neighborhood from a fixed 50-node + high-degree seed set. GFQL: `[n(is_in seeds), e_undirected(hops=k), n()]`. + GraphFrames has no k-hop-neighborhood primitive (`bfs` = shortest path + between predicates, `find` = fixed motif), so we expand via iterated + undirected edge joins — still pure Spark. +- **pagerank** — full-graph PageRank. GFQL CPU/polars → + `g.compute_igraph('pagerank')`; GPU → `g.compute_cugraph('pagerank')`. + GraphFrames → `gf.pageRank(resetProbability=0.15, maxIter=N)` (damping 0.85). + +## Why median-of-5 + 2 warmups + +Warmups absorb one-time costs — JIT, lazy-frame plan compilation, Spark JVM +class-loading and executor spin-up, filesystem cache priming — so timed runs +measure steady-state compute, not startup. Median (not mean) of 5 is robust to +the occasional GC pause / stop-the-world spike on a shared box. Cold load is +timed separately, once, because it is a different question (ETL cost) from +warm query latency. + +## Fairness caveats (documented, not hidden) + +- **Spark JVM warmup**: even after 2 warmups, `local[*]` carries per-query + scheduler/task-serialization overhead that dominates on small results — + Spark is built for distributed throughput, not single-node latency. +- **Materialization**: Spark is lazy; every task ends in `.count()` (or + `.vertices.count()`) to force honest end-to-end timing. GFQL likewise + materializes via `len(_nodes)/len(_edges)`. +- **`local[*]` vs distributed**: this measures single-box multicore, GraphFrames' + single-node configuration, not a distributed cluster. A real cluster would amortize overhead differently; + we are explicitly benchmarking the single-node regime where GFQL lives. +- **GFQL PageRank routing**: polars has no native PageRank; the polars engine + converts to pandas and calls igraph. That conversion is inside the timed + region (honest), but it means "gfql-polars pagerank" ≈ igraph-on-CPU. +- **PageRank iterations**: GraphFrames uses fixed `maxIter=N`; igraph/cugraph + iterate to a tolerance. Not iteration-for-iteration identical — compare + wall-clock-to-usable-scores, not per-iteration cost. + +## Parity (same answer on both) + +Each task returns a `result_size` written to JSONL: filter → node count above +threshold, hop → neighborhood size, pagerank → vertex count. Filter and hop +sizes should match exactly across systems (identical set semantics); a mismatch +flags a bug (e.g. directedness or seed-set drift). The harness validates +successful rows after the sweep and exits nonzero if any dataset/task has a +`result_size` mismatch across successful systems. PageRank scores are compared by +rank correlation (Spearman) of the top-K vertices offline, not exact values, +since the algorithms differ in convergence criteria. + +## Guardrails + +Every (system, task) is wrapped: an error/OOM records +`{"status":"error","error":...}` and the run continues. Missing pyspark / +graphframes / GPU is skipped with a message — never aborts the matrix. +Results stream to JSONL (one line per system×task×dataset), flushed per row so +a mid-run crash still leaves partial data. +``` +python bench_graphframes.py --dataset lj --systems gfql-polars,graphframes \ + --tasks filter,hop1,hop2,pagerank --warmups 2 --iters 5 --out lj.jsonl +``` +``` +python bench_graphframes.py --dataset orkut --dry-run # print plan only +``` diff --git a/benchmarks/gfql/bench_ladybug.py b/benchmarks/gfql/bench_ladybug.py new file mode 100644 index 0000000000..e4d16164c0 --- /dev/null +++ b/benchmarks/gfql/bench_ladybug.py @@ -0,0 +1,252 @@ +#!/usr/bin/env python3 +""" +bench_ladybug.py — GFQL-native port of the LadybugDB vs Kuzu benchmark. + +Mirrors https://github.com/LadybugDB/kuzu-ladybug-benchmark (synthetic Item/Owns +graph) so GFQL (polars CPU, and cudf GPU on a GPU host) can be compared +head-to-head against LadybugDB 0.18.0 / Kuzu 0.11.3 on the SAME query shapes. + +The Ladybug suite's operations map onto GFQL/dataframe primitives (see the +per-op comments). Several are direct analogues of our own work: + - op9 out-degree for seeded nodes == our CSR ``edge_out_adj`` seeded index + - op11 scan-rel rowid == columnar edge scan / Arrow return + - op13 Arrow CSR export == our ``create_index('edge_out_adj')`` CSR + +SAFETY: default size is TINY (1K nodes / 5K edges) and engine is ``polars`` +(CPU) so this can run locally for correctness validation without a GPU and +without large memory. Use ``--nodes 5M --edges 20M`` (matching Ladybug's full +run) and ``--engine cudf`` ONLY on a GPU host with headroom. + +Usage: + python bench_ladybug.py # tiny local validation (CPU polars) + python bench_ladybug.py --validate # + assert results vs pandas oracle + python bench_ladybug.py -n 5M -e 20M # full scale (run on the bench box) + python bench_ladybug.py --engine cudf -n 5M -e 20M # GPU (bench box only) +""" +from __future__ import annotations + +import argparse +import gc +import json +import os +import statistics +import sys +import time +from typing import Any, Callable, Dict, List, Optional, Tuple + +# Prefer the working-tree graphistry over any (possibly stale) pip-installed one, +# so submodules like graphistry.compute.chain resolve to this repo's code. +_REPO = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) +if os.path.isdir(os.path.join(_REPO, "graphistry")) and _REPO not in sys.path: + sys.path.insert(0, _REPO) + + +def parse_num(s: str) -> int: + s = str(s).upper().strip() + if s.endswith("M"): + return int(float(s[:-1]) * 1_000_000) + if s.endswith("K"): + return int(float(s[:-1]) * 1_000) + return int(s) + + +def build_dataset(n_nodes: int, n_edges: int): + """Replicate Ladybug's synthetic Item/Owns graph as pandas frames. + + nodes: id 1..N, name 'abcdefghijklmn_name_{id}' + edges: i 1..E -> src=(i%N)+1, dst=(i*7%N)+1, since=i (deterministic) + """ + import numpy as np + import pandas as pd + + ids = np.arange(1, n_nodes + 1, dtype="int64") + nodes = pd.DataFrame({"id": ids, "name": [f"abcdefghijklmn_name_{i}" for i in ids]}) + i = np.arange(1, n_edges + 1, dtype="int64") + src = (i % n_nodes) + 1 + dst = ((i * 7) % n_nodes) + 1 + edges = pd.DataFrame({"src": src, "dst": dst, "since": i}) + return nodes, edges + + +def timed_median(fn: Callable[[], Any], warmups: int, iters: int) -> Tuple[float, Optional[int]]: + for _ in range(warmups): + fn() + gc.collect() + samples: List[float] = [] + last: Optional[int] = None + for _ in range(iters): + t0 = time.perf_counter() + out = fn() + samples.append((time.perf_counter() - t0) * 1000.0) + last = out if isinstance(out, int) else (len(out) if hasattr(out, "__len__") else None) + gc.collect() + return statistics.median(samples), last + + +class GFQLLadybug: + """GFQL adapter running the Ladybug op-suite. engine in {'polars','cudf'}.""" + + def __init__(self, engine: str, nodes, edges): + import graphistry + + self.engine = engine + self.nodes = nodes + self.edges = edges + self.n_nodes = int(len(nodes)) + self.n_edges = int(len(edges)) + self.g = graphistry.edges(edges, "src", "dst").nodes(nodes, "id") + + # -- op fns: each returns a size (int) for a cheap correctness signal ------ + def op_full_scan(self): # Ladybug #4 + g, eng = self.g, self.engine + from graphistry.compute.ast import n + return lambda: int(len(g.gfql([n()], engine=eng)._nodes)) + + def op_range(self, lo: int, hi: int): # Ladybug #5 <-- the index question + # Columnar range predicate. Hypothesis: a vectorized full-column + # scan is already competitive without a dedicated range index. + g, eng = self.g, self.engine + from graphistry.compute.ast import n + from graphistry.compute.predicates.numeric import between + return lambda: int(len(g.gfql([n({"id": between(lo, hi)})], engine=eng)._nodes)) + + def op_point(self, pid: int): # Ladybug #6 + g, eng = self.g, self.engine + from graphistry.compute.ast import n + return lambda: int(len(g.gfql([n({"id": pid})], engine=eng)._nodes)) + + def op_count_rel(self): # Ladybug #8 + edges = self.edges + return lambda: int(len(edges)) + + def op_out_degree_seeded(self, k: int = 100): # Ladybug #9 == our CSR seeded index + # Vectorized dataframe-native out-degree for nodes id 1..k (GFQL's + # strength vs a per-node query loop). Sum of out-degrees. + edges = self.edges + seeds = set(range(1, k + 1)) + + def run() -> int: + deg = edges.groupby("src").size() + return int(sum(int(deg.get(s, 0)) for s in seeds)) + + return run + + def op_scan_rel(self): # Ladybug #10 + edges = self.edges + return lambda: int(len(edges[["src", "dst", "since"]])) + + def op_scan_rel_rowid(self): # Ladybug #11 (their 50-60x claim vs Kuzu) + edges = self.edges + return lambda: int(len(edges[["src", "dst"]])) + + def op_arrow_csr(self): # Ladybug #13 == our create_index('edge_out_adj') CSR + g = self.g + + def run() -> int: + gi = g.create_index("edge_out_adj") + # size signal: number of indexes present after build + try: + idx = gi.show_indexes() + return int(len(idx)) if hasattr(idx, "__len__") else 1 + except Exception: + return 1 + + return run + + +# (name, method, args, engine_exec): engine_exec=False ops run on the raw pandas +# ingest frames regardless of --engine (dataframe-level baselines) and are labeled +# system='gfql-pandas-df' so an --engine cudf run never reports pandas timings +# under a GPU label. Engine-native Cypher timings live in bench_ladybug_cypher.py. +OPS = [ + ("full_scan", "op_full_scan", (), True), + ("range", "op_range", None, True), # args filled from size + ("point", "op_point", None, True), + ("count_rel", "op_count_rel", (), False), + ("out_degree_seeded", "op_out_degree_seeded", (), False), + ("scan_rel", "op_scan_rel", (), False), + ("scan_rel_rowid", "op_scan_rel_rowid", (), False), + ("arrow_csr", "op_arrow_csr", (), True), +] + + +def main(argv: Optional[List[str]] = None) -> int: + p = argparse.ArgumentParser(description="GFQL-native LadybugDB benchmark suite") + p.add_argument("-n", "--nodes", default="1K") + p.add_argument("-e", "--edges", default="5K") + p.add_argument("--engine", default="polars", choices=["polars", "cudf", "pandas"]) + p.add_argument("--warmups", type=int, default=1) + p.add_argument("--iters", type=int, default=3) + p.add_argument("--out", default=None) + p.add_argument("--validate", action="store_true", + help="assert op results vs a pandas oracle (tiny sizes)") + p.add_argument("--debug", action="store_true", help="print tracebacks on op errors") + args = p.parse_args(argv) + + n_nodes, n_edges = parse_num(args.nodes), parse_num(args.edges) + # SAFETY rail for accidental local big runs. + if args.engine != "cudf" and (n_nodes > 2_000_000 or n_edges > 8_000_000): + print(f"[GUARD] {n_nodes:,} nodes / {n_edges:,} edges is large for a CPU " + f"host — run this on the bench box. Refusing by default.", file=sys.stderr) + return 3 + + print(f"dataset: {n_nodes:,} nodes / {n_edges:,} edges, engine={args.engine}", file=sys.stderr) + nodes, edges = build_dataset(n_nodes, n_edges) + lo, hi = n_nodes // 2, n_nodes // 2 + 1000 + mid = n_nodes // 2 + + adapter = GFQLLadybug(args.engine, nodes, edges) + + # Fill parametrized op args. + resolved = [] + for name, meth, a, engine_exec in OPS: + if name == "range": + a = (lo, hi) + elif name == "point": + a = (mid,) + resolved.append((name, meth, a, engine_exec)) + + expected: dict = {} + if args.validate: + # pandas oracle: a timing is void unless the op's result size matches + expected = { + "full_scan": n_nodes, + "range": int(((nodes["id"] >= lo) & (nodes["id"] <= hi)).sum()), + "point": int((nodes["id"] == mid).sum()), + "count_rel": n_edges, + "scan_rel": n_edges, + "scan_rel_rowid": n_edges, + } + print(f"[oracle] {expected}", file=sys.stderr) + + results = [] + for name, meth, a, engine_exec in resolved: + system = f"gfql-{args.engine}" if engine_exec else "gfql-pandas-df" + fn = getattr(adapter, meth)(*a) + try: + med, size = timed_median(fn, args.warmups, args.iters) + if args.validate and name in expected and size != expected[name]: + raise AssertionError( + f"oracle mismatch: {name} size={size} expected={expected[name]} — timing void") + row = {"system": system, "op": name, "n_nodes": n_nodes, + "n_edges": n_edges, "median_ms": round(med, 3), "size": size, "status": "ok"} + print(f"[OK] {name:20} median={med:8.3f}ms size={size}", file=sys.stderr) + except Exception as ex: + import traceback as _tb + row = {"system": system, "op": name, "status": "error", + "error": str(ex)[:300]} + print(f"[ERR] {name}: {ex}", file=sys.stderr) + if args.debug: + _tb.print_exc() + results.append(row) + + if args.out: + with open(args.out, "w") as fh: + for r in results: + fh.write(json.dumps(r) + "\n") + print(f"wrote {args.out}", file=sys.stderr) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/benchmarks/gfql/bench_ladybug_cypher.py b/benchmarks/gfql/bench_ladybug_cypher.py new file mode 100644 index 0000000000..c5327d5c96 --- /dev/null +++ b/benchmarks/gfql/bench_ladybug_cypher.py @@ -0,0 +1,96 @@ +#!/usr/bin/env python3 +"""FAIR GFQL-vs-Ladybug: run Ladybug's benchmark ops as GFQL Cypher MATCH...RETURN +(the row pipeline), NOT dataframe shortcuts. Compares against LadybugDB's PUBLISHED +5M/20M numbers (their figures, their hardware — a cross-machine comparison; treat +ratios as indicative). Ladybug best (5M/20M, ms): full_scan 3789, range 7.5, +point 0.3, count 3.3, out_degree100 59.8, scan_rel_props 15722, +scan_rel_rowid 14562. Kuzu count 46. + +FAIRNESS: each engine is benchmarked on a graph built in ITS OWN NATIVE frame type +(pandas/polars/cuDF), built ONCE outside the timing loop. An earlier version built the +graph in pandas for ALL engines, so every `engine='polars'/'cudf'` call re-converted the +5M-row (string-column) frame — ~200ms of pandas->polars conversion that swamped sub-10ms +queries and made polars/cudf look 27-675x slower than they are. A real polars user keeps +data in polars (as Ladybug keeps its own native store); native-per-engine is the honest +comparison. +""" +import os, time, statistics, sys +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))) +import numpy as np, pandas as pd, graphistry + +N = int(os.environ.get("N", "5000000")); E = int(os.environ.get("E", "20000000")) +REPS = int(os.environ.get("REPS", "5")); WARM = int(os.environ.get("WARM", "2")) +ENGINES = os.environ.get("ENGINES", "pandas,polars").split(",") +mid = N // 2; lo, hi = N // 2, N // 2 + 1000 + +def build(engine): + """Build the graph in ``engine``'s native frame type (no per-call conversion).""" + ids = np.arange(1, N + 1, dtype=np.int64) + name = ("abcdefghijklmn_name_" + pd.Series(ids).astype(str)).values + i = np.arange(1, E + 1, dtype=np.int64) + src = (i % N) + 1; dst = ((i * 7) % N) + 1; since = i + node_cols = {"id": ids, "name": name} + edge_cols = {"src": src, "dst": dst, "since": since} + if engine in ("polars", "polars-gpu"): + import polars as pl + nd, ed = pl.DataFrame(node_cols), pl.DataFrame(edge_cols) + elif engine == "cudf": + import cudf + nd, ed = cudf.DataFrame(node_cols), cudf.DataFrame(edge_cols) + else: # pandas + nd, ed = pd.DataFrame(node_cols), pd.DataFrame(edge_cols) + return graphistry.nodes(nd, "id").edges(ed, "src", "dst") + +OPS = { + "full_scan": "MATCH (i) RETURN i.id, i.name", + "range": f"MATCH (i) WHERE i.id >= {lo} AND i.id <= {hi} RETURN i.id, i.name", + "point": f"MATCH (i) WHERE i.id = {mid} RETURN i.id, i.name", + "count": "MATCH ()-[r]->() RETURN COUNT(*)", + "scan_rel_props": "MATCH (a)-[o]->(b) RETURN a.id, b.id, o.since", + "scan_rel_rowid": "MATCH (a)-[r]->(b) RETURN a.id, b.id", +} +LADYBUG = {"full_scan": 3789, "range": 7.5, "point": 0.3, "count": 3.3, + "scan_rel_props": 15722, "scan_rel_rowid": 14562} + +def _size(res): + df = getattr(res, "_nodes", None) + return None if df is None else int(len(df)) + +def med(fn): + """Warm median; returns (ms, result_size) — a timing is void unless the size is + consistent across reps (and callers compare it across engines).""" + for _ in range(WARM): fn() + t, sizes = [], set() + for _ in range(REPS): + s = time.perf_counter(); r = fn(); t.append((time.perf_counter() - s) * 1e3) + sizes.add(_size(r)) + assert len(sizes) == 1, f"unstable result size across reps: {sizes} — timing void" + return statistics.median(sorted(t)), sizes.pop() + +def main(): + print(f"N={N:,} E={E:,} REPS={REPS} (native-per-engine, built once)", flush=True) + print(f"{'op':16} {'engine':11} {'gfql_ms':>10} {'ladybug_ms':>11} {'vs_ladybug':>11}") + # Build each engine's native graph ONCE, then run all its ops (no per-call conversion). + op_sizes: dict = {} # op -> size from the first engine; later engines must match + for eng in ENGINES: + try: + g = build(eng) + except Exception as ex: + print(f"{'(build)':16} {eng:11} {'SKIP':>10} -> {type(ex).__name__}: {str(ex)[:40]}", flush=True) + continue + for name, q in OPS.items(): + try: + ms, size = med(lambda: g.gfql(q, engine=eng)) + if name in op_sizes and size != op_sizes[name]: + print(f"{name:16} {eng:11} {'VOID':>10} -> size {size} != {op_sizes[name]} (cross-engine mismatch)", flush=True) + continue + op_sizes.setdefault(name, size) + lb = LADYBUG.get(name) + vs = f"{lb/ms:.1f}x" if lb else "-" + print(f"{name:16} {eng:11} {ms:10.3f} {str(lb):>11} {vs:>11} rows={size}", flush=True) + except Exception as ex: + print(f"{name:16} {eng:11} {'NIE/ERR':>10} -> {type(ex).__name__}: {str(ex)[:40]}", flush=True) + del g + +if __name__ == "__main__": + main() diff --git a/docs/source/gfql/_static/graphframes/bench_graphframes_pagerank_parity.json b/docs/source/gfql/_static/graphframes/bench_graphframes_pagerank_parity.json new file mode 100644 index 0000000000..a04c670327 --- /dev/null +++ b/docs/source/gfql/_static/graphframes/bench_graphframes_pagerank_parity.json @@ -0,0 +1,15 @@ +{ + "n_common_vertices": 3997962, + "spearman": { + "igraph_vs_cugraph": 1.0, + "igraph_vs_graphframes": 1.0, + "cugraph_vs_graphframes": 1.0 + }, + "top100_overlap": { + "igraph_vs_cugraph": 100, + "igraph_vs_graphframes": 100, + "cugraph_vs_graphframes": 100 + }, + "dataset": "lj", + "note": "PageRank score agreement across engines; GraphFrames maxIter=20, igraph eps=1e-3, cugraph tol=1e-5" +} \ No newline at end of file diff --git a/docs/source/gfql/_static/graphframes/results.json b/docs/source/gfql/_static/graphframes/results.json new file mode 100644 index 0000000000..e794ad8f2c --- /dev/null +++ b/docs/source/gfql/_static/graphframes/results.json @@ -0,0 +1,198 @@ +{ + "lj": { + "n_edges": 34681189, + "n_nodes": 3997962, + "tasks": { + "filter": { + "gfql-polars": { + "cold_load_ms": 2362.8, + "iters": 5, + "median_ms": 2.1, + "result_size": 403561, + "warmups": 2 + }, + "gfql-polars-gpu": { + "cold_load_ms": 2290.0, + "iters": 5, + "median_ms": 2.4, + "result_size": 403561, + "warmups": 2 + }, + "graphframes": { + "cold_load_ms": 10255.6, + "iters": 5, + "median_ms": 90.4, + "result_size": 403561, + "warmups": 2 + } + }, + "hop1": { + "gfql-polars": { + "cold_load_ms": 2362.8, + "iters": 5, + "median_ms": 236.8, + "result_size": 119877, + "warmups": 2 + }, + "gfql-polars-gpu": { + "cold_load_ms": 2290.0, + "iters": 5, + "median_ms": 191.4, + "result_size": 119877, + "warmups": 2 + }, + "graphframes": { + "cold_load_ms": 10255.6, + "iters": 5, + "median_ms": 1421.7, + "result_size": 119877, + "warmups": 2 + } + }, + "hop2": { + "gfql-polars": { + "cold_load_ms": 2362.8, + "iters": 5, + "median_ms": 1669.3, + "result_size": 1378430, + "warmups": 2 + }, + "gfql-polars-gpu": { + "cold_load_ms": 2290.0, + "iters": 5, + "median_ms": 1542.1, + "result_size": 1378430, + "warmups": 2 + }, + "graphframes": { + "cold_load_ms": 10255.6, + "iters": 5, + "median_ms": 3583.3, + "result_size": 1378430, + "warmups": 2 + } + }, + "pagerank": { + "gfql-polars": { + "cold_load_ms": 2362.8, + "iters": 5, + "median_ms": 49307.6, + "result_size": 3997962, + "warmups": 2 + }, + "gfql-polars-gpu": { + "cold_load_ms": 2783.1, + "iters": 3, + "median_ms": 1110.9, + "result_size": 3997962, + "warmups": 1 + }, + "graphframes": { + "cold_load_ms": 10255.6, + "iters": 5, + "median_ms": 16336.0, + "result_size": 3997962, + "warmups": 2 + } + } + } + }, + "orkut": { + "n_edges": 117185083, + "n_nodes": 3072441, + "tasks": { + "filter": { + "gfql-polars": { + "cold_load_ms": 5145.3, + "iters": 5, + "median_ms": 1.7, + "result_size": 308666, + "warmups": 2 + }, + "gfql-polars-gpu": { + "cold_load_ms": 4916.8, + "iters": 5, + "median_ms": 2.0, + "result_size": 308666, + "warmups": 2 + }, + "graphframes": { + "cold_load_ms": 14725.6, + "iters": 5, + "median_ms": 70.6, + "result_size": 308666, + "warmups": 2 + } + }, + "hop1": { + "gfql-polars": { + "cold_load_ms": 5145.3, + "iters": 5, + "median_ms": 562.9, + "result_size": 434973, + "warmups": 2 + }, + "gfql-polars-gpu": { + "cold_load_ms": 4916.8, + "iters": 5, + "median_ms": 442.0, + "result_size": 434973, + "warmups": 2 + }, + "graphframes": { + "cold_load_ms": 14725.6, + "iters": 5, + "median_ms": 3826.6, + "result_size": 434973, + "warmups": 2 + } + }, + "hop2": { + "gfql-polars": { + "cold_load_ms": 5145.3, + "iters": 5, + "median_ms": 9439.8, + "result_size": 1991366, + "warmups": 2 + }, + "gfql-polars-gpu": { + "cold_load_ms": 4916.8, + "iters": 5, + "median_ms": 8860.2, + "result_size": 1991366, + "warmups": 2 + }, + "graphframes": { + "cold_load_ms": 14725.6, + "iters": 5, + "median_ms": 11582.9, + "result_size": 1991366, + "warmups": 2 + } + }, + "pagerank": { + "gfql-polars": { + "cold_load_ms": 5145.3, + "iters": 5, + "median_ms": 160097.1, + "result_size": 3072441, + "warmups": 2 + }, + "gfql-polars-gpu": { + "cold_load_ms": 4916.8, + "iters": 5, + "median_ms": 3502.8, + "result_size": 3072441, + "warmups": 2 + }, + "graphframes": { + "cold_load_ms": 14725.6, + "iters": 5, + "median_ms": 36826.0, + "result_size": 3072441, + "warmups": 2 + } + } + } + } +} \ No newline at end of file diff --git a/docs/source/gfql/benchmark_graphframes.rst b/docs/source/gfql/benchmark_graphframes.rst new file mode 100644 index 0000000000..5d431ff89b --- /dev/null +++ b/docs/source/gfql/benchmark_graphframes.rst @@ -0,0 +1,413 @@ +GFQL Graph Benchmark: DataFrame-Native vs Apache Spark GraphFrames +================================================================== + +.. image:: _static/gfql-mascot.png + :alt: GFQL mascot + :width: 160px + :align: right + +.. note:: + + LiveJournal and Orkut figures are final: median of 5 timed runs after 2 + warmups, result-size parity enforced per task. One cell — LiveJournal GPU + PageRank — is median of 3 after 1 warmup (a re-run after a transient GPU + fault on the first pass); every other cell, including Orkut GPU PageRank, is + the full 5/2. Friendster (~1.8B edges) was the stretch target; our *eager + in-memory* harness runs out of RAM loading it (documented below) — this is a + harness/loader limit, not an engine ceiling. Polars' streaming engine and the + cudf-polars streaming executor are the larger-than-memory paths, not yet + benchmarked here. + +Run graph filters, k-hop neighborhoods, and PageRank directly on Python +dataframes — no cluster required. This benchmark compares **GFQL** +(Graphistry's dataframe-native graph query language) on CPU +(``engine="polars"``) and GPU (``engine="polars-gpu"``) against **Apache Spark +GraphFrames** (``local[*]``, single-node JVM) on the same tasks over large +SNAP graphs. + +The short version: for **filter and traversal**, GFQL wins decisively — even on +CPU — because a single-node columnar engine avoids the JVM startup, +task-serialization, and shuffle overhead that dominate Spark at sub-second +result sizes. For **PageRank**, the honest answer is mixed: GFQL's *CPU* path +routes through igraph and is *slower* than GraphFrames at scale; GFQL's win on +PageRank comes from the *GPU* path (cugraph). We state both plainly below. + +Headline (LiveJournal, ~35M edges) +---------------------------------- + +.. list-table:: + :header-rows: 1 + :widths: 26 18 18 18 20 + + * - Task + - GFQL polars (CPU) + - GFQL polars-gpu (GPU) + - GraphFrames (local[*]) + - Best GFQL vs GraphFrames + * - **filter** (degree >= 42) + - 2.1ms + - 2.4ms + - 90.4ms + - **~43x** + * - **1-hop** (50 seeds) + - 236.8ms + - 191.4ms + - 1421.7ms + - **~7.4x** + * - **2-hop** (50 seeds) + - 1669.3ms + - 1542.1ms + - 3583.3ms + - **~2.3x** + * - **PageRank** (full graph) + - 49.3s + - **1.11s** + - 16.3s + - **~14.7x** (GPU) / *0.33x* (CPU) + +*Median of 5 after 2 warmups (LiveJournal GPU PageRank is median of 3 — see the +note above). DGX* ``dgx-spark``, *GB10 GPU, single node; Spark* ``local[*]`` +*over all cores. Cold load (ETL) of the SNAP file is 2.4s for GFQL vs 10.3s for +GraphFrames — GFQL also loads ~4x faster.* + +Result-size parity is enforced per task: filter +returns the identical node count above threshold, 1-hop the identical +neighborhood size (**119,877**), 2-hop the identical size (**1,378,430**), and +PageRank the identical vertex count (**3,997,962**). A size mismatch flags a bug +(directedness or seed-set drift), not a speedup. + +When GFQL wins, and when it doesn't +----------------------------------- + +This page is written for a Spark GraphFrames user evaluating alternatives. +The point is not to spin — it is to be trustworthy. Two findings, both true: + +**1. Filter and traversal: GFQL wins across the board (1.3–43x; most cells 2x+), even on CPU.** +There is no JVM to warm, no task graph to serialize, no shuffle to schedule. A +single-node columnar engine is simply the right tool for sub-second graph +queries. Spark's ``local[*]`` per-query scheduler overhead dominates at these +result sizes — Spark is engineered for distributed throughput across a cluster, +not single-node latency. Note the GPU barely moves these numbers: at this scale +the CPU polars path is already fast enough that data movement, not compute, is +the floor. + +**2. PageRank: the honest result is mixed — reach for the GPU.** +GFQL's *CPU* path has no native PageRank, so the polars engine converts to +pandas and calls igraph. Single-threaded igraph is **slower than GraphFrames** +at this scale (49.3s vs 16.3s on LiveJournal, and 160s vs 37s on Orkut — the gap +widens with size): Spark's multicore iterative aggregation genuinely beats it. +GFQL's PageRank advantage comes entirely from the **GPU** path (cugraph, +~1.11s), which beats GraphFrames by ~14.7x. So the +guidance is explicit: for whole-graph analytics like PageRank, use the GPU +engine; the CPU-igraph route is a convenience, not a speed play. + +If you take one thing away: **GFQL replaces Spark for interactive single-node +graph queries, and the GPU engine additionally replaces it for whole-graph +analytics — but the CPU engine alone does not win PageRank, and we won't +pretend it does.** + +filter — WHERE on a degree column +--------------------------------- + +A ``WHERE`` on a numeric column: keep nodes with ``degree >= threshold``. SNAP +graphs carry no attributes, so ``degree`` is precomputed at cold-load (charged +to load, not to the query, for *both* systems) and used as the natural +threshold column. + +.. doc-test: skip + +.. code-block:: python + + # GFQL + from graphistry import n + from graphistry.compute.predicates.numeric import ge + g.gfql([n(filter_dict={'degree': ge(42)})], engine="polars") # or "polars-gpu" + + # GraphFrames + gf.degrees.filter("degree >= 42").count() + +LiveJournal: GFQL polars **2.1ms**, GFQL polars-gpu **2.4ms**, GraphFrames +**90.4ms** — same node count (**403,561**) on a shared degree threshold. The gap +is almost entirely Spark's per-query scheduling floor; the actual predicate is +trivial on both. + +1-hop — neighborhood from a 50-node seed set +-------------------------------------------- + +Undirected 1-hop expansion from a fixed 50-node high-degree seed set. + +.. doc-test: skip + +.. code-block:: python + + # GFQL + from graphistry import n, e_undirected + g.gfql([n(filter_dict={'id': is_in(seeds)}), e_undirected(hops=1), n()], engine="polars") + +GraphFrames has no k-hop-neighborhood primitive (``bfs`` is shortest-path +between predicates, ``find`` is a fixed motif), so the Spark side expands via an +iterated undirected edge join — still pure Spark, ending in ``.count()``. + +LiveJournal: GFQL polars **236.8ms**, GFQL polars-gpu **191.4ms**, GraphFrames +**1421.7ms**, identical neighborhood size **119,877**. + +2-hop — two-hop neighborhood +---------------------------- + +Same seed set, two undirected hops (``e_undirected(hops=2)`` for GFQL; two +iterated joins for Spark). + +LiveJournal: GFQL polars **1669.3ms**, GFQL polars-gpu **1542.1ms**, GraphFrames +**3583.3ms**, identical size **1,378,430**. As the result grows, real join work +starts to dominate Spark's fixed overhead, so the multiple narrows (~2.3x) — but +GFQL still wins on a single node. + +PageRank — full-graph analytics +------------------------------- + +Full-graph PageRank (damping 0.85). GFQL CPU routes to igraph +(``g.compute_igraph('pagerank')``); GFQL GPU routes to cugraph +(``g.compute_cugraph('pagerank')``); GraphFrames uses +``gf.pageRank(resetProbability=0.15, maxIter=20)``. GraphFrames runs a fixed +20 iterations; igraph and cugraph iterate to their library-default tolerance +(igraph ``eps=1e-3``, cugraph ``tol=1e-5``). This favors neither side +uniformly — it is disclosed so the times are interpretable, not a hidden knob. + +LiveJournal (all return **3,997,962** vertices): + +.. list-table:: + :header-rows: 1 + :widths: 40 30 30 + + * - Engine / backend + - Time + - vs GraphFrames + * - GFQL polars / igraph (CPU) + - 49.3s + - *0.33x (slower)* + * - GFQL polars-gpu / cugraph (GPU) + - **1.11s** + - **~14.7x faster** + * - GraphFrames (local[*]) + - 16.3s + - 1.0x + +This is the mixed result, stated plainly. The CPU-igraph route is single +threaded and **loses to Spark's multicore aggregation** here. The GPU-cugraph +route wins by an order of magnitude. Because GraphFrames uses a fixed +``maxIter`` while igraph/cugraph iterate to a tolerance, the raw scores are not +bit-identical, so we compare **wall-clock-to-usable-scores**: the three engines +return the identical vertex set (**3,997,962**), and their PageRank rankings +agree **exactly** — pairwise Spearman rho = **1.00** and top-100 overlap +**100/100** across igraph, cugraph, and GraphFrames (parity check saved to +``bench_graphframes_pagerank_parity.json``). This is a "same ranked result, different cost" comparison, not a raced approximation. + +Orkut (~117M edges) +------------------- + +.. list-table:: + :header-rows: 1 + :widths: 26 18 18 18 20 + + * - Task + - GFQL polars (CPU) + - GFQL polars-gpu (GPU) + - GraphFrames (local[*]) + - Best GFQL vs GraphFrames + * - **filter** (degree >= 162) + - 1.7ms + - 2.0ms + - 70.6ms + - **~42x** + * - **1-hop** (50 seeds) + - 562.9ms + - 442.0ms + - 3826.6ms + - **~8.7x** + * - **2-hop** (50 seeds) + - 9439.8ms + - 8860.2ms + - 11582.9ms + - **~1.3x** + * - **PageRank** (full graph) + - 160.1s + - **3.50s** + - 36.8s + - **~10.5x** (GPU) / *0.23x* (CPU) + +*Median of 5 after 2 warmups (all cells, including GPU PageRank). +Result-size parity per task: filter* **308,666**; *1-hop* **434,973**; *2-hop* +**1,991,366**; *PageRank* **3,072,441**. *Cold load 5.1s (GFQL) vs 14.7s +(GraphFrames). The pattern holds at 117M edges: GFQL wins filter/traversal +outright, the GPU wins PageRank by ~10x, and CPU-igraph PageRank falls further +behind Spark (0.23x) as the graph grows.* + +Friendster (~1.8B edges) — our eager-load harness stops here; streaming is next +-------------------------------------------------------------------------------- + +Friendster (1,806,067,135 edges, 65.6M nodes) was the stretch target. Every path +we *ran* ran out of headroom on the **119 GB** node — but the honest framing is +that this is where **our benchmark harness's eager, in-memory load** stops, **not +a hard ceiling of the engines.** The harness reads the whole graph into memory up +front (``pandas.read_parquet`` → a ~29 GB edge frame, plus a second ~29 GB pass to +build the degree/node table) *before the query runs*; that materialization is what +the OS kills. + +.. list-table:: + :header-rows: 1 + :widths: 26 74 + + * - Path (as configured in this harness) + - Outcome at 1.8B edges on one 119 GB node + * - GFQL polars (CPU), eager load + - **OOM in the load**, before the query: the pandas edge frame + degree build + peak past physical RAM. The *query* engine never runs. + * - GFQL polars-gpu (GPU), eager cudf load + - **Exceeds memory in the load**: even a lean cudf-direct edge read drives the + 119 GB unified pool into swap. The in-memory GPU executor is not the + larger-than-memory path (see below). + * - GraphFrames (local[*]) + - **Swap-thrash.** A ``local[*]`` driver with a 90 GB heap on a 1.8B-edge + GraphFrame saturates memory and does not finish in usable time on one box. + +**What we did *not* run — the larger-than-memory paths that exist.** GFQL's Polars +engine already ships opt-in streaming escape hatches, and this harness did not use +them: + +- **CPU:** ``GFQL_POLARS_CPU_STREAMING=1`` collects the plan with Polars' streaming + engine (batched, spills to disk), parity-identical to the default. Paired with a + **lazy** source (``pl.scan_parquet`` instead of an eager ``pandas.read_parquet``), + the 1.8B-edge input is never fully materialized. +- **GPU:** ``GFQL_POLARS_GPU_EXECUTOR=streaming`` selects the cudf-polars *streaming* + executor — explicitly the escape hatch for **larger-than-device-memory** results, + where the default in-memory executor would OOM. + +Both are **off by default** because in-scope GFQL graphs/results fit in memory and +streaming regresses small/interactive sizes — the right default for the 35M–117M +regime this page measures. What we have *not yet* done is wire a lazy +``scan_parquet`` ingestion path through GFQL and benchmark the streaming collect at +1.8B; that is the correct larger-than-memory test (comparable to Ladybug's +out-of-core mode and to a Spark cluster) and is **tracked as follow-up work**, not a +limitation we're conceding. So: GFQL wins decisively *in-memory* through ~10^8 edges +here; at ~10^9 the question is streaming-vs-out-of-core-vs-cluster, which we will +measure rather than assert. + +Why this matters +---------------- + +Most graph work in a notebook or a pipeline is single-node and latency +sensitive: filter to a subgraph, expand a few hops, score it. For that regime, +standing up or paying for a Spark cluster is the wrong shape — the per-query +scheduling and serialization cost swamps the actual work. GFQL runs the same +queries in-process on your dataframe, on CPU, and wins by 1.3–43x here +(most cells 2x+; the closest is Orkut's heavy 2-hop at 1.3x). + +When the workload shifts to whole-graph analytics like PageRank, the GPU engine +(``engine="polars-gpu"``, cugraph) is the tool that beats Spark — by ~10–15x +(14.7x on LiveJournal, 10.5x on Orkut) — on the same single node. The CPU +engine's PageRank is a convenience for when no GPU is present, not a performance +claim. + +**When to go back to Spark.** These *in-memory* numbers hold while the graph and +its intermediates fit in one machine's memory (here, 119 GB unified host/GPU +memory comfortably holds Orkut's 117M edges). Above that, GFQL has two moves +before a cluster: Polars' **streaming engine** (``GFQL_POLARS_CPU_STREAMING=1``, +disk-spill) and the **cudf-polars streaming executor** +(``GFQL_POLARS_GPU_EXECUTOR=streaming``, larger-than-device-memory) — both +opt-in, both untested at 1.8B here (see the Friendster section). A managed Spark +cluster is the right tool when the data already lives there, or when the graph +outgrows even streaming on one node. This page measures the in-memory single-node +regime; it does not claim GFQL replaces a cluster at every scale, nor that +one node is a hard ceiling. + +Fairness and caveats (documented, not hidden) +--------------------------------------------- + +We benchmark the single-node regime where GFQL lives, and we flag every place +that favors or disfavors either side: + +- **local[*] is Spark's single-node configuration.** This measures single-box + multicore, not a distributed cluster. A real cluster amortizes scheduling and + shuffle overhead across many machines and would change the trade-off, + especially at larger scales. We are explicitly benchmarking single-node + latency, which is where GFQL is designed to run. +- **End-to-end materialization on both sides.** Spark is lazy, so every task + ends in a materializing action (``.count()`` / ``.vertices.count()``) to force + honest end-to-end timing. GFQL likewise materializes via + ``len(_nodes)`` / ``len(_edges)``. Both are timed to a real answer, not a lazy + plan. +- **The pandas→polars conversion is charged to GFQL.** GFQL holds edges as + pandas and converts to polars *inside* the timed region on each call. This is + conservative — it counts against GFQL — and is left in deliberately rather + than pre-converting. +- **PageRank convergence differs (disclosed).** GraphFrames runs a fixed + ``maxIter=20``; igraph iterates to ``eps=1e-3`` and cugraph to ``tol=1e-5``. + The comparison is wall-clock-to-usable-scores; we verify all three return the + identical vertex set and rank it identically (LiveJournal: pairwise Spearman + rho = 1.00, top-100 overlap 100/100 — ``bench_graphframes_pagerank_parity.json``), + not per-iteration cost — the algorithms converge to the same ranking at + different cost. +- **In-memory by default (streaming is opt-in).** These results are the default + *in-memory* configuration, which assumes the graph fits in one node's RAM — the + regime this page measures. GFQL does **not** shard across machines, but it *can* + spill to disk / stream: Polars' streaming engine + (``GFQL_POLARS_CPU_STREAMING=1``) and the cudf-polars streaming executor + (``GFQL_POLARS_GPU_EXECUTOR=streaming``) are larger-than-memory paths, off by + default and not exercised in these numbers. We report the host memory (119 GB) + so the in-memory envelope is explicit. +- **Runs are blocked, not interleaved.** On this shared box, GFQL and + GraphFrames were run in separate blocks (all GFQL cells, then all GraphFrames + cells), not interleaved, and only medians are retained per cell. Validation and + final medians agreed within run-to-run noise; we report medians, consistent + with :doc:`benchmark_filter_pagerank`. +- **Warmups and median.** 2 warmups absorb one-time costs (JIT, lazy-plan + compilation, JVM class-loading, executor spin-up, filesystem cache priming) so + the timed runs measure steady state. Median of 5 (not mean) is robust to the + occasional GC / stop-the-world spike on a shared box. Cold load (ETL) is timed + separately, once — a different question from warm query latency. +- **Guardrails.** Each (system, task) is wrapped: an error/OOM records a status + and the matrix continues; missing pyspark/graphframes/GPU is skipped with a + message, never aborting the run. + +Reproducibility +--------------- + +Results are rendered from saved JSON (``_static/graphframes/results.json``) — +this page does **not** rerun benchmarks. The committed harness is +``benchmarks/gfql/bench_graphframes.py`` (design notes in +``benchmarks/gfql/bench_graphframes_DESIGN.md``). To reproduce the LiveJournal +matrix (from ``benchmarks/gfql/``, with the graphframes jar on the Spark +classpath via ``GRAPHFRAMES_JAR``): + +.. code-block:: bash + + python bench_graphframes.py --dataset lj \ + --systems gfql-polars,gfql-polars-gpu,graphframes \ + --tasks filter,hop1,hop2,pagerank \ + --filter-threshold 42 --warmups 2 --iters 5 + +Orkut uses ``--dataset orkut --filter-threshold 162``. The shared +``--filter-threshold`` makes the filter task bit-identical across systems. + +Environment +----------- + +- Host: ``dgx-spark``, single node; GPU: ``GB10`` +- GFQL engines: ``engine="polars"`` (CPU, PageRank via igraph) and + ``engine="polars-gpu"`` (GPU, PageRank via cugraph) +- Spark: GraphFrames ``0.8.4-spark3.5-s_2.12``, PySpark ``3.5.1``, ``local[*]`` +- Datasets: `SNAP `_ LiveJournal (~35M edges), + Orkut (~117M edges), Friendster (~1.8B edges, stretch) +- Measurement: median of 5 runs after 2 warmups; result-size parity enforced + per task; results rendered from saved JSON + +See also +-------- + +- :doc:`engines` — choosing an engine; four-engine and external-tool comparison + (including where PuppyGraph / warehouse-federated tools fit — not yet + benchmarked head-to-head) +- :doc:`benchmark_filter_pagerank` — GFQL CPU/GPU vs Neo4j + GDS +- :doc:`cypher` — Cypher syntax through ``g.gfql("MATCH ...")`` +- :doc:`overview` — GFQL design, features, and GPU acceleration +- :doc:`about` — 10-minute introduction to GFQL diff --git a/docs/source/gfql/engines.rst b/docs/source/gfql/engines.rst index ecdf680ea1..4b7c169724 100644 --- a/docs/source/gfql/engines.rst +++ b/docs/source/gfql/engines.rst @@ -198,6 +198,32 @@ benchmarked** rather than guess. seeds): **22× Kuzu**, up to **87× at k=100k**. See :doc:`index_adjacency`. - **Not claimed:** cyclic / multi-way-join patterns (triangles, cliques) where Kuzu's worst-case-optimal joins can win. Use Kuzu as the store; GFQL for bulk read analytics. + * - **LadybugDB** + - Actively-maintained **Kuzu fork** (Kuzu is archived); embedded C++, strongly-typed + Cypher, opt-in ART *or* hash indexing, zero-copy Arrow/CSR scans, and **out-of-core + billion-scale** (query a 1.8B-edge graph in <8 GB RAM). + - Against **LadybugDB's published numbers** for their own 5M-node / 20M-edge suite + (their figures, their hardware; GFQL measured separately on an NVIDIA DGX Spark + GB10 running the identical Cypher ``MATCH … RETURN`` row pipeline, each engine on + its **native** frames — a cross-machine comparison, so read the ratios as + indicative): GFQL **wins the scan-shaped ops** — full node scan **~65×** (polars + 58 ms vs 3789 ms), id **range ~1.2×** (polars 6.1 ms vs 7.5 ms), relationship + property/rowid scans **~3.5–3.7×** (cuDF 4.2 s vs ~15 s). **Point lookup** (single + id) is ~4 ms vs Ladybug's ~0.3 ms — a full columnar scan vs a B-tree/hash **index + seek**; close in absolute terms, and a resident GFQL node-id index (tracked in + issue #1676) should close it. Ladybug still wins the two ops backed by + persistent structure: point lookups and a relationship ``COUNT(*)`` (an O(1) cached + count vs GFQL's O(E) endpoint-validated scan — a dataframe has no referential + integrity). GFQL's angle is dataframe-native, in-process, and GPU-accelerated with + no separate store to load/index. + - **Complement:** Ladybug is a durable embedded store with an out-of-core mode + (billion-scale in <8 GB RAM); GFQL is a query engine over your dataframes. GFQL's + *default* is in-memory, but it is **not limited to it** — Polars streaming + (``GFQL_POLARS_CPU_STREAMING=1``, disk-spill) and the cudf-polars streaming executor + (``GFQL_POLARS_GPU_EXECUTOR=streaming``) are larger-than-memory paths + (billion-scale head-to-head not yet benchmarked — see :doc:`benchmark_graphframes`). + Natural split: Ladybug as the persistent/out-of-core store; pull a subgraph into GFQL + for GPU analytics — or run GFQL streaming directly on your columnar files. * - **igraph** - Pure-Python/C graph library. - — (not a standalone competitor here) @@ -211,9 +237,11 @@ benchmarked** rather than guess. * - **Spark GraphFrames** - *Distributed* graph engine on a Spark cluster; provision + tune the cluster. - GFQL is *single-node* (CPU or one GPU): 100M+ edges in-process on **one machine**, - no cluster to stand up, interactive latency — and a single GPU often matches or beats - a Spark cluster on read-heavy traversal + PageRank at a fraction of the cost. - *Head-to-head not yet published.* + no cluster to stand up, interactive latency — and a single node often matches or beats + Spark on read-heavy traversal and, with the GPU engine, PageRank at a fraction of the cost. + Head-to-head on LiveJournal (35M) and Orkut (117M): GFQL wins filter/traversal 1.3–43× + even on CPU, and the GPU engine wins PageRank ~10–15×; on CPU, PageRank via igraph is + *slower* than GraphFrames — see :doc:`benchmark_graphframes`. - Reach for GraphFrames when the graph genuinely exceeds one machine's memory. Motif / triangle / multi-way-join queries **run** in GFQL but are not yet perf-benchmarked. * - **PuppyGraph** @@ -552,7 +580,7 @@ Parity and honesty Methodology ----------- -- Host: ``dgx-spark`` (GB10 Grace-Blackwell, unified memory — the F3 memory-pressure +- Host: NVIDIA DGX Spark (GB10 Grace-Blackwell, unified memory — the F3 memory-pressure boundary is partly a property of this box), RAPIDS container ``graphistry/test-rapids-official:26.02-gfql-polars``. - Datasets: `SNAP `_ **com-LiveJournal** (35M edges), @@ -564,6 +592,10 @@ Methodology - Reproduce: ``benchmarks/gfql/index_bulk_olap_bench.py`` (engine comparison), ``benchmarks/gfql/pandas_vs_polars.py``, and ``benchmarks/gfql/index_vs_kuzu_prepared.py`` (vs kuzu). Numbers on this page are rendered from saved runs; the page does not re-run them. +- **LadybugDB row**: the Ladybug figures are **their published results on their hardware**; + the GFQL side ran on the host above via ``benchmarks/gfql/bench_ladybug_cypher.py`` + (5M/20M synthetic per their suite shape, native frames per engine, warm medians) — a + cross-machine comparison, disclosed as such in the row. Install ------- diff --git a/docs/source/gfql/index.rst b/docs/source/gfql/index.rst index 7ac79d738a..fdf8aac350 100644 --- a/docs/source/gfql/index.rst +++ b/docs/source/gfql/index.rst @@ -56,6 +56,7 @@ See also: Seeded Traversal Indexes GFQL CPU & GPU Acceleration End-to-End Benchmark + vs Spark GraphFrames translate combo quick