neo vs Spark.

elif args.lib == "spark":
    t0 = now()
    from pyspark.sql import SparkSession, functions as F
    import pyspark
    res["version"] = pyspark.__version__
    mem = os.environ.get("NEON_SPARK_MEMORY", "48g")
    spark = (SparkSession.builder.master(f"local[{args.threads}]")
             .config("spark.driver.memory", mem)
             .config("spark.sql.shuffle.partitions", str(args.threads))
             .config("spark.default.parallelism", str(args.threads))
             .config("spark.ui.enabled", "false")
             .config("spark.sql.adaptive.enabled", "true")
             .getOrCreate())
    spark.sparkContext.setLogLevel("ERROR")
    res["startup_ms"] = (now() - t0) * 1000
    PIDS = descendants()  # the JVM is where Spark's work is done
    res["java"] = spark.sparkContext._jvm.java.lang.System.getProperty("java.version")

    df = None

    def load():
        global df
        if df is not None:
            df.unpersist(blocking=True)
        df = spark.read.csv(args.csv, header=True, inferSchema=True).cache()
        df.count()  # force the read into memory; inferSchema has already read the file once more for types
    res["load_first_ms"], res["load_ms"] = load_twice(load)
    res["rows"] = df.count()
    pred = F.col("amount") > 500
    sink = lambda d: d.write.format("noop").mode("overwrite").save()  # Spark's benchmarking sink: every row is produced, none is collected
    timed("filter", lambda: df.filter(pred).count())
    timed("filter_mat", lambda: sink(df.filter(pred)))
    res["sort_perm_ms"] = None
    timed("sort_mat", lambda: sink(df.orderBy("amount")))
    timed("agg", lambda: df.groupBy("region").agg(F.sum("amount")).collect())
    checks["filter_count"] = df.filter(pred).count()
    checks["first_sorted"] = [float(r[0]) for r in df.orderBy("amount").select("amount").limit(3).collect()]
    checks["sums"] = {r[0]: float(r[1]) for r in df.groupBy("region").agg(F.sum("amount")).collect()}
    energy_summary()  # while the JVM is still there to be read
    spark.stop()  # so the JVM's peak RSS is in RUSAGE_CHILDREN

 194  l.t = csv::parse(raw);
 337  const double f5 = timeIt([&] { Bitmap b = compareDoubles(amt.data(), n, Cmp::Gt, 500, &pool); sink = b.words[0]; }, 7);
 372  const double s4 = timeIt([&] { sortIndexByDouble(amt.data(), n, perm, &pool); sink = perm[0]; }, 3);
 454  const double a5 = G <= 16 ? timeIt([&] { groupSumAnyOrder(amtP, idP, n, G, &pool, pairTotal); sink = static_cast<size_t>(pairTotal[0]); }, 9) : 0;

// energy.h — what a process has used, as the operating system's own ledger has it: the energy it was billed,
// its processor time, the wake-ups it caused. The one reader of that ledger, for two questions:
//   what a kernel costs — the Evals page's Energy view (bench_engine_options.cpp reads it around each kernel;
//     tools/energy/energy.cpp checks the counter and reads it for a command it runs;
//     bench/lang/python/yardstick.py reads the same field; PHD.md section 33)
//   what a process costs at rest — the idle studies (scripts/idle_cost.sh and scripts/idle_server.py take
//     `neon_energy --pid <pid>` twice and subtract; PHD.md section 31.1)
//
// macOS on Apple silicon: proc_pid_rusage(RUSAGE_INFO_V6). ri_energy_nj — nanojoules billed to the process's
// threads for the processor cores they ran on, since it started. Not the memory, the disk or the screen, and
// not the machine at rest: what running this code added. (Not ri_billed_energy: against half a second of a
// busy loop that field read 0.08 mJ where ri_energy_nj read 1,887.) Beside it, processor time in nanoseconds
// — `ps` has hundredths of a second, more than a resting window uses in a minute — and the wake-ups. No
// privilege is needed, and any process of the same user can be read, so a tree of processes can be summed.
// Linux: /proc/<pid>/stat and /proc/<pid>/status — clock ticks and context switches. No energy: RAPL is one
// figure for the whole machine and root's to read.
// Elsewhere: nothing is read (ok stays false).
//
// A reading is cumulative; a measurement is the difference of two. The counter moves in steps (a thread's
// share is settled when it leaves a core or enters the kernel), so a window should be a quarter of a second
// or more: `neon_energy --check` prints how far apart the steps are on the machine it runs on.