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 전문 보기 ↗