This crate contains benchmarks based on popular public data sets and open source benchmark suites, making it easy to run real-world benchmarks to help with performance and scalability testing and for comparing performance with other Arrow implementations as well as other query engines.
These benchmarks are derived from the TPC-H benchmark.
TPC-H data can be generated using tpchgen-rs, a fast TPC-H data generator written in Rust.
Install via pip:
pip install tpchgen-cli
Or via cargo:
cargo install tpchgen-cli
Generate SF=1 data in Parquet format:
tpchgen-cli -s 1 --format parquet --output-dir data
For larger scale factors (e.g., SF=10):
tpchgen-cli -s 10 --format parquet --output-dir data
Data will be generated into the data subdirectory and will not be checked in because this directory has been added to the .gitignore file.
Build the Python bindings and then run:
$ python tpch.py --query q1 --path /mnt/bigdata/tpch/sf1-parquet/ Registering table part at path /mnt/bigdata/tpch/sf1-parquet//part Registering table supplier at path /mnt/bigdata/tpch/sf1-parquet//supplier Registering table partsupp at path /mnt/bigdata/tpch/sf1-parquet//partsupp Registering table customer at path /mnt/bigdata/tpch/sf1-parquet//customer Registering table orders at path /mnt/bigdata/tpch/sf1-parquet//orders Registering table lineitem at path /mnt/bigdata/tpch/sf1-parquet//lineitem Registering table nation at path /mnt/bigdata/tpch/sf1-parquet//nation Registering table region at path /mnt/bigdata/tpch/sf1-parquet//region Query q1 took 9.668351173400879 second(s)
Note that this Python script currently only supports running against file formats than contain a schema definition (such as Parquet).
The benchmark can then be run (assuming the data created from dbgen is in ./data) with a command such as:
cargo run --release --bin tpch -- benchmark datafusion --iterations 3 --path ./data --format tbl --query 1 --batch-size 4096
You can enable the feature mimalloc (to use the mimalloc allocator) as features by passing them in as --features:
cargo run --release --features "mimalloc" --bin tpch -- benchmark datafusion --iterations 3 --path ./data --format tbl --query 1 --batch-size 4096
The benchmark program also supports CSV and Parquet input file formats and a utility is provided to convert from tbl (generated by the dbgen utility) to CSV and Parquet.
cargo run --release --bin tpch -- convert --input ./data --output /mnt/tpch-parquet --format parquet
This utility does not yet provide support for changing the number of partitions when performing the conversion. Another option is to use the following Docker image to perform the conversion from tbl files to CSV or Parquet.
docker run -it ballistacompute/spark-benchmarks:0.4.0-SNAPSHOT -h, --help Show help message Subcommand: convert-tpch -i, --input <arg> --input-format <arg> -o, --output <arg> --output-format <arg> -p, --partitions <arg> -h, --help Show help message
Note that it is necessary to mount volumes into the Docker container as appropriate so that the file conversion process can access files on the host system.
Here is a full example that assumes that data is stored in the /mnt path on the host system.
docker run -v /mnt:/mnt -it ballistacompute/spark-benchmarks:0.4.0-SNAPSHOT \ convert-tpch \ --input /mnt/tpch/csv \ --input-format tbl \ --output /mnt/tpch/parquet \ --output-format parquet \ --partitions 64
To run the benchmarks it is necessary to have at least one Ballista scheduler and one Ballista executor running.
To run the scheduler from source:
cd $ARROW_HOME/ballista/scheduler RUST_LOG=info cargo run --release
By default the scheduler will bind to 0.0.0.0 and listen on port 50050.
To run the executor from source:
cd $ARROW_HOME/ballista/executor RUST_LOG=info cargo run --release
By default the executor will bind to 0.0.0.0 and listen on port 50051.
You can add mimalloc/LTO flags to improve speed (with longer build times):
RUST_LOG=info RUSTFLAGS='-C target-cpu=native -C lto -C codegen-units=1 -C embed-bitcode' cargo run --release --bin executor --features "mimalloc" --target x86_64-unknown-linux-gnu
To run the benchmarks:
cd $ARROW_HOME/benchmarks cargo run --release --bin tpch benchmark ballista --host localhost --port 50050 --query 1 --path $(pwd)/data --format tbl
Pass --output <dir> (works for both the ballista and datafusion benchmark subcommands) to write a machine-readable summary to <dir>/tpch-<start_time>.json. Each run records the benchmark and DataFusion versions, CPU count, the full CLI arguments (so the configuration is self-describing), and per-query per-iteration elapsed times and row counts.
The summary is rewritten after every query, atomically, so a run that is killed part way through (for example an out-of-memory SIGKILL at a large scale factor) still leaves the results collected so far on disk. A query that fails is recorded with an error message and its completed iterations, and the run continues to the remaining queries instead of aborting; the process still exits non-zero if any query failed.
Compare two summary files query-by-query — fastest iteration per side, delta, delta percent, suite totals, and row-count agreement:
cargo run --release --bin tpch compare baseline.json candidate.json
The docker-compose.yml at the repo root brings up one scheduler and two executors (8 vCPU / 8 GB memory pool each) plus a benchmark client. End to end, including data generation, image builds, and running all 22 TPC-H queries:
./dev/integration-tests.sh
Defaults to SF=10, 16 partitioned Parquet files per table, 3 iterations per query. Override with env vars:
SCALE_FACTOR=1 PARTITIONS=8 ITERATIONS=1 ./dev/integration-tests.sh
The script generates Parquet via tpchgen-cli (installed via cargo install if missing), builds the host binaries (dev/build-ballista-executables.sh), builds the docker images (dev/build-ballista-docker.sh), waits for healthy services, runs the queries, and tears down the stack on exit (success or failure).
To run pieces by hand:
SCALE_FACTOR=1 ./benchmarks/tpch-gen.sh # generate Parquet under benchmarks/data ./dev/build-ballista-executables.sh # cargo build the binaries on the host ./dev/build-ballista-docker.sh # build docker images that COPY the binaries docker compose up -d --wait # bring up scheduler + 2 executors + client docker compose run --rm ballista-client /root/run.sh docker compose down --remove-orphans
Note:
dev/build-ballista-executables.shbuilds the binaries on the host, so the host needs a Rust toolchain that targets Linux (matching the runtime images). Linux hosts work directly; macOS/Windows users need to either build on a Linux host or arrange cross-compilation.
The result of query 1 should produce the following output when executed against the SF=1 dataset.
+--------------+--------------+----------+--------------------+--------------------+--------------------+--------------------+--------------------+----------------------+-------------+ | l_returnflag | l_linestatus | sum_qty | sum_base_price | sum_disc_price | sum_charge | avg_qty | avg_price | avg_disc | count_order | +--------------+--------------+----------+--------------------+--------------------+--------------------+--------------------+--------------------+----------------------+-------------+ | A | F | 37734107 | 56586554400.73001 | 53758257134.870026 | 55909065222.82768 | 25.522005853257337 | 38273.12973462168 | 0.049985295838396455 | 1478493 | | N | F | 991417 | 1487504710.3799996 | 1413082168.0541 | 1469649223.1943746 | 25.516471920522985 | 38284.467760848296 | 0.05009342667421622 | 38854 | | N | O | 74476023 | 111701708529.50996 | 106118209986.10472 | 110367023144.56622 | 25.502229680934594 | 38249.1238377803 | 0.049996589476752576 | 2920373 | | R | F | 37719753 | 56568041380.90001 | 53741292684.60399 | 55889619119.83194 | 25.50579361269077 | 38250.854626099666 | 0.05000940583012587 | 1478870 | +--------------+--------------+----------+--------------------+--------------------+--------------------+--------------------+--------------------+----------------------+-------------+ Query 1 iteration 0 took 1956.1 ms Query 1 avg time: 1956.11 ms
We run benchmarks to compare performance with Spark to identify future areas of optimization. We publish the latest results in the top-level README.
Build with the release-lto profile.
cargo build --profile release-lto
Run the cluster.
./target/release-lto/ballista-scheduler ./target/release-lto/ballista-executor -c 24
Running the benchmark.
./target/release-lto/tpch benchmark ballista \ --host localhost \ --port 50050 \ --path /mnt/bigdata/tpch/sf10-parquet-float/ \ --format parquet \ --iterations 1 \ --partitions 24 \ --query 1
Start the cluster.
./sbin/start-master.sh ./sbin/start-worker.sh spark://ripper:7077
Run the benchmark.
$SPARK_HOME/bin/spark-submit \ --master spark://ripper:7077 \ --class org.apache.arrow.ballista.SparkTpch \ --conf spark.driver.memory=8G \ --num-executors=1 \ --conf spark.executor.memory=32G \ --conf spark.executor.cores=24 \ --conf spark.cores.max=24 \ target/spark-tpch-0.5.0-SNAPSHOT-jar-with-dependencies.jar \ tpch \ --input-path /mnt/bigdata/tpch/sf10-parquet-float/ \ --input-format parquet \ --query-path /home/andy/git/apache/datafusion-ballista/benchmarks/queries \ --query 1
cargo run --bin tpch -- loadtest ballista-load --query-list 1,3,5,6,7,10,12,13 --requests 200 --concurrency 10 --data-path /**** --format parquet --host localhost --port 50050 --sql-path /*** --debug
Unlike the TPC-H suite above (which measures performance), the TPC-DS suite is a correctness gate: it runs each query on a Ballista cluster and compares the result, row-by-row, against a single-process DataFusion oracle running the same query.
The 99 TPC-DS queries are checked in at benchmarks/queries-tpcds/qN.sql, so no fetch step is needed to run the harness. They were vendored from DataFusion's branch-54 (the queries DataFusion uses in its own TPC-DS tests) with:
./dev/vendor-tpcds-queries.sh
That script fetches 1.sql .. 99.sql from datafusion/core/tests/tpc-ds on the configured branch (DATAFUSION_BRANCH, default branch-54) and writes them to benchmarks/queries-tpcds/qN.sql (prefixed with q to match the TPC-H naming convention). Re-run it only to refresh the committed queries when the DataFusion pin changes.
TPC-DS data is generated with the same tpchgen-rs project as TPC-H, via its tpcgen-cli tpcds subcommand. TPC-DS support is not yet published to crates.io, so the script installs tpcgen-cli from git, pinned to a fixed rev:
SCALE_FACTOR=1 OUTPUT_DIR=./data-tpcds ./benchmarks/tpcds-gen.sh
SCALE_FACTOR (default 1) and OUTPUT_DIR (default benchmarks/data-tpcds, resolved relative to the script) are env overrides; TPCGEN_REV overrides the pinned tpcgen-cli git rev if needed. Note that, unlike the TPC-H generator, tpcgen-cli tpcds has no --parts flag — it writes a single <table>.parquet file per table for all 24 TPC-DS tables.
Bring up a scheduler and executor as described above in “Running the Ballista Benchmarks”, then run the tpcds binary with --verify:
cargo run --release --bin tpcds -- \ --host localhost --port 50050 \ --path $(pwd)/data-tpcds \ --partitions 16 \ --verify \ -c datafusion.optimizer.prefer_hash_join=false
With no --query given, this runs every non-skipped query (see “Skip list” below) on the Ballista cluster and, because --verify is set, also runs it against a single-process DataFusion SessionContext and diffs the results. The process exits non-zero and prints a summary if any query fails or mismatches. --verify assumes a local --path: the oracle context registers the tables directly and does not apply object-store credentials.
Some queries are excluded from the gate. These are listed in the SKIP const near the top of benchmarks/src/bin/tpcds.rs, as (query_id, reason) pairs, grouped by cause: queries whose distributed result diverges from DataFusion (tracked as bugs), queries that are non-deterministic under LIMIT/ORDER BY ties, and queries whose tpcgen-cli schema column names differ from the DataFusion query text. A default run (no --query) executes 1..=99 minus SKIP.
.github/workflows/tpcds.yml runs the SF1 suite against a local scheduler + executor on every push/PR touching ballista/** or benchmarks/**, under the default (static) planner. The adaptive planner (AQE on) is not gated yet — it currently fails many TPC-DS queries with an EmptyExec invalid partition assertion (issue #2047); re-enable an AQE-on run once that is fixed.