andygrove commented on code in PR #2161: URL: https://github.com/apache/datafusion-ballista/pull/2161#discussion_r3641780182
########## docs/source/contributors-guide/benchmarking.md: ########## @@ -19,384 +19,165 @@ # Benchmarking -This page describes how Ballista is benchmarked at scale, and records the current -results. It is aimed at contributors who want to reproduce a number, understand why -a comparison is set up the way it is, or add a result of their own. +Current TPC-H **SF1000** results for Ballista, compared against a vanilla +**Spark 3.4** baseline running on the same cluster shape. -The benchmark used here is derived from TPC-H. For running TPC-H locally at small -scale factors, see [`benchmarks/README.md`][bench-readme] in the repository; this -page covers the multi-node, large-scale-factor setup and the cross-engine -comparisons. +## Versions under test -[bench-readme]: https://github.com/apache/datafusion-ballista/blob/main/benchmarks/README.md - -## What we measure, and why - -Ballista's central performance question is **how evenly work is spread across -executor tasks**. A distributed query is only as fast as its slowest stage, and a -stage is only as fast as its slowest task, so a query whose work concentrates onto -a few partitions will underperform no matter how fast the underlying operators are. -Benchmarks are therefore run at a scale where that imbalance actually shows up. - -These benchmarks are run with **adaptive query execution (AQE) enabled** -(`ballista.planner.adaptive.enabled=true`). This is the configuration Ballista is -being actively developed against, so it is the one measured here. - -AQE selects the adaptive planner, which can re-plan stages using runtime statistics — -coalescing partitions, re-optimising joins, and promoting a small join side to a -broadcast at runtime. The alternative, AQE off, selects the static -`DefaultDistributedPlanner`; it is a different planner with materially different join -behaviour, and it is no longer benchmarked on this page. +| Engine | Version | +| -------- | -------------------------------------------------------------------------------------------------------------------------------- | +| Ballista | [`696ca29b`](https://github.com/apache/datafusion-ballista/commit/696ca29b3c1fe3013238822b7f5d8cdb918fdaa3) (`main`, 2026-07-23), Cargo pkg `54.0.0`, DataFusion `54.1.0` | +| Spark | 3.4 (vanilla, no acceleration plugin) | ## Environment -The results on this page currently come from a **small homelab cluster**: two -bare-metal nodes running Kubernetes, with the TPC-H data staged on node-local disk. -This is a deliberate starting point rather than a final destination. A two-node -cluster with local disk removes as many confounds as possible — no object-store -latency, no cloud network variance, no noisy neighbours — so that when a query is -slow, the cause is Ballista and not the environment. - -The intent is to **move these benchmarks to AWS with data in S3** once the results -here are good. That environment is the one users actually run, and it exercises -things a homelab cannot: object-store reads instead of local disk, higher and more -variable network latency between executors, and larger executor counts. Some of -Ballista's behaviour is expected to change there — a shuffle that is cheap over a -local link is not cheap over a cloud network, and object-store reads make scan -parallelism matter differently. - -So results on this page should be read as **relative comparisons on controlled -hardware**, useful for "did this change help", not as absolute throughput numbers -for a cloud deployment. - -## Reference cluster - -All results on this page use the following shape. It is a reference point, not a -requirement — the commands below work on any cluster, but numbers are only -comparable when the shape matches. - -| | | -| ------------------- | ------------------------------------------- | -| Executors | 2, one per physical node | -| Per executor | 16 cores, 56 GiB, `--memory-pool-size=48GB` | -| Per task slot | 16 concurrent tasks → 3 GB pool each | -| Scheduler | 1 | -| Data | TPC-H SF1000 Parquet, node-local disk | -| `target_partitions` | 64 | - -Two details matter more than they look: - -- **Executors are spread one per node.** Packing two executors onto one node makes - them contend for the same disk and memory bandwidth, which is measuring the host, - not the engine. -- **Data is on node-local disk, not object storage.** With a modest network between - nodes, reading from object storage makes the interconnect the bottleneck and the - engine comparison becomes an I/O comparison. - -The memory pool is deliberately set below the container limit. Ballista splits -`--memory-pool-size` into one `FairSpillPool` per task slot, and that accounting -does not cover every allocation, so leaving headroom between the pool and the -container limit avoids the container being killed outright instead of reporting a -graceful resource error. +- **Cluster:** Kubernetes on AWS (`us-west-2`); one driver/scheduler pod and + 32 executor pods for each engine, launched on the same node pool. +- **Executor pod (Ballista):** x86_64, 8 vCPU, 32 GiB memory. +- **Executor pod (Spark):** x86_64, 8 vCPU, 64 GiB + 10 GiB overhead. +- **Data:** TPC-H SF1000 Parquet on S3 (`us-west-2`), ZSTD compression, + ~512 MiB row groups, one directory per table. + +## Ballista configuration + +| Flag / config key | Value | +| ------------------------------------------------------------- | -------------- | +| `--concurrent-tasks` | `8` | +| `--memory-pool-size` (bytes; ≈70 % of the 32 GiB container) | `24051816858` | +| `datafusion.execution.target_partitions` | `256` | +| `datafusion.execution.collect_statistics` | `true` | +| `datafusion.execution.listing_table_factory_infer_partitions` | `false` | +| `datafusion.catalog.information_schema` | `true` | +| `ballista.planner.adaptive.enabled` | `true` (AQE) | +| `ballista.shuffle.sort_based.memory_limit_per_task_bytes` | `0` | + +`datafusion.optimizer.prefer_hash_join` is left at its default; under AQE the +join strategy is selected at runtime by `DelayJoinSelectionRule` / +`DynamicJoinSelectionExec` from runtime statistics and the broadcast / +`ballista.optimizer.hash_join_max_build_partition_bytes` thresholds. + +## Spark configuration (highlights) + +Vanilla Spark 3.4 — no Comet plugin, stock `SortShuffleManager`. + +| Key | Value | +| ---------------------------------- | ------------------------------------------- | +| `spark.executor.instances` | `32` | +| `spark.executor.cores` | `16` (task parallelism per executor) | +| `spark.kubernetes.executor.limit.cores` / `.request.cores` | `8` (physical vCPU) | +| `spark.executor.memory` | `64G` | +| `spark.executor.memoryOverhead` | `10G` | +| `spark.memory.fraction` | `0.6` | +| `spark.memory.storageFraction` | `0.2` | +| `spark.sql.shuffle.partitions` | `512` | +| `spark.sql.broadcastTimeout` | `900` | +| `spark.serializer` | `KryoSerializer` | +| `spark.io.compression.codec` | `zstd` | + +Spark AQE is left at its Spark 3.4 defaults. Shuffle spills to a `gp3`-backed +per-executor volume (`spark.kubernetes.executor.volumes...spark-local-dir-1`). ## Queries -All engines run the **same SQL**, which is what makes a cross-engine comparison -apples-to-apples. +The SQLBench-H phrasing of the 22 TPC-H queries from +[apache/datafusion-benchmarks](https://github.com/apache/datafusion-benchmarks). -The shared set is the TPC-H queries from [apache/datafusion-benchmarks][dfb] -(SQLBench-H). Ballista's bundled queries in `benchmarks/queries/` are the classic -TPC-H phrasing and differ in places, so the shared set is overlaid over them for a -comparison run. - -Spark and Comet are driven through Comet's benchmark harness, which reads its own -bundled copy of the queries (labelled CometBench-H). That copy is textually -identical to the SQLBench-H set — all 22 queries match once the licence header -comment is ignored — so the engines are executing the same statements even though -they load them from different paths. Worth re-checking if either set is ever -regenerated. - -[dfb]: https://github.com/apache/datafusion-benchmarks +## Results -## Running the benchmark +**Times in seconds; lower is better.** Ballista: mean of 2 iterations. Spark: +mean of 3 iterations (cold iteration dropped by the harness). `FAIL` = query +did not complete on this Ballista configuration. + +| Query | Ballista (s) | Spark 3.4 (s) | +| ----: | -----------: | ------------: | +| 1 | 19.55 | 67.58 | +| 2 | 32.84 | 29.80 | +| 3 | 40.52 | 25.13 | +| 4 | 27.98 | 21.19 | +| 5 | 202.84 | 54.12 | Review Comment: I'll need to look into this. I filed https://github.com/apache/datafusion-ballista/issues/2162 so that I don't lose track of this -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
