Ballista: Making DataFusion Applications Distributed
Ballista is a distributed query execution engine that enhances Apache DataFusion
by enabling the parallelized execution of workloads across multiple nodes in a distributed environment.
It is also a lighter, Rust-native alternative to Apache Spark, keeping the same execution
model of stages, shuffles and adaptive query execution. On a 1 TB benchmark derived from TPC-H, Ballista is
2.2x faster than Spark by default and 3.4x faster with experimental pipelined execution.
An existing DataFusion application can be distributed with a few lines of code changed:
+use ballista::prelude::*;
use datafusion::prelude::*;
#[tokio::main]
async fn main() -> datafusion::error::Result<()> {
- let ctx = SessionContext::new();
+ // create SessionContext with ballista support
+ // standalone context will start all required
+ // ballista infrastructure in the background as well
+ let ctx = SessionContext::standalone().await?;
// register the table
ctx.register_csv("example", "tests/data/example.csv", CsvReadOptions::new())
.await?;
// create a plan to run a SQL query
let df = ctx
.sql("SELECT a, MIN(b) FROM example WHERE a <= b GROUP BY a LIMIT 100")
.await?;
// execute and print results
df.show().await?;
Ok(())
}
For documentation or more examples, please refer to the Ballista User Guide.
Performance
Ballista is measured against Apache Spark 4.1.3 on the same cluster shape: Kubernetes on AWS, with 32 executors of
8 vCPUs running the 22 queries over 1 TB of Parquet in S3. Each Ballista executor had 64 GiB of memory, and each Spark
executor 74 GiB.
Derived from TPC-H and not an official TPC-H result. Times are the mean of two runs, including the cold first run. The
Spark numbers come from an earlier run with separate scripts and its own phrasing of the queries. See the
benchmarking guide for versions, settings, the full results and
how to reproduce them.
Features
Query interfaces
- SQL and DataFrames from Rust and Python, using DataFusion's APIs
- Arrow Flight SQL, so ADBC and JDBC drivers and BI tools can query a cluster without linking Ballista
- SQL CLI with a built-in terminal UI
- Substrait and physical plans accepted alongside logical plans
DataFrame.cache() keeps intermediate results on the executors for reuse within a session (opt-in)
Execution
- Adaptive query execution, on by default, re-optimizes the remaining plan from runtime statistics: join selection,
broadcast joins, empty-stage elimination and exchange reuse, with opt-in skew-join splitting and partition coalescing
- Two execution modes that share one cluster: fault-tolerant staged execution by default, for batch and ETL jobs,
and low-latency pipelined execution (experimental), for interactive queries
- Memory-bounded executors with an auto-sized pool by default, and spilling to disk
- Sort-based and hash shuffle with configurable compression
- Multi-partition tasks that share memory and broadcast state across a stage's partitions
- Parallel window functions over range shuffles (opt-in)
Operations
- Deployment as native binaries or Docker images, with Docker Compose and Kubernetes guides and optional KEDA
autoscaling
- Fault tolerance: failed tasks are retried, and work lost with an executor is re-run
- Graceful shutdown and Kubernetes health probes
- Observability: a browser-based Web TUI, a terminal UI, a REST API with an OpenAPI spec, Prometheus metrics, and a
history server for completed jobs
- Result service for clients outside the cluster network
- Data access to Parquet, CSV and JSON on S3-compatible object stores, HTTP and local filesystems
Extensibility
- Reusable components: embed the scheduler and executor crates in your own engine
- Extension points for custom operators and codecs, functions, object stores, session configuration and adaptive
optimizer rules
Who is Ballista for
Ballista serves several distinct audiences:
以上为 README 的开头部分。其余内容请阅读原文。 查看完整 README ↗
