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:

• DataFusion users going multi-node — you already use Apache DataFusion on a single machine and have outgrown it. Ballista runs the same SQL and DataFrame workloads across a cluster with minimal code changes and the same results.
• Spark users wanting the same execution model — you run Spark SQL or batch jobs and want a lighter, Rust-native alternative without relearning a new paradigm. Ballista keeps the familiar model: plans split into stages at shuffle boundaries, tasks over partitions, executors with vcores, and adaptive query execution (AQE).
• Library users building a specialized engine — you are building a bespoke distributed query engine and want reusable scheduler, executor, and plan-serialization building blocks with extension points, instead of writing distributed execution from scratch.

These audiences are documented in more detail, along with the guarantees each relies on, in the User Personas guide.

Architecture

A Ballista cluster consists of one or more scheduler processes and one or more executor processes. These processes can be run as native binaries and are also available as Docker Images, which can be easily deployed with Docker Compose or Kubernetes.

Clients submit query plans to the scheduler over gRPC, or SQL over Arrow Flight SQL. The scheduler breaks each query into stages and sends tasks to the executors, which report progress back. Executors exchange shuffle data with each other over Arrow Flight and return results to the client the same way.

See the architecture guide for more details.

Execution Modes

Ballista supports two execution modes, selected with ballista.execution.mode. Staged execution, the default, runs a job one stage at a time, the way Apache Spark does: each stage writes its output to shuffle files, and a failed task is simply re-run from them, which suits long-running jobs and busy shared clusters.

Pipelined execution (experimental) runs a job the way Trino does: every task is scheduled at once and stages stream record batches to each other over Arrow Flight, with no intermediate data written to disk, so results reach the client sooner, at the cost of a failed task failing the whole job. It suits short, latency-sensitive queries on clusters with room for the whole job. See the Pipelined Execution Guide for details.

One Engine for ETL and Interactive Queries

A lakehouse aims to merge the data warehouse and the ETL platform, but it is often still served by two kinds of engine: a batch engine for long ETL jobs, built to survive task failures, and an interactive engine for short analytical queries, built for low latency. Ballista's two execution modes let one engine serve both, from the same scheduler and executors. Long ETL jobs run staged, where a failed task is re-run from shuffle files. Short analytical queries run pipelined, where results stream to the client as they are produced. A session picks its mode and can change it between queries, so the same SQL and DataFrame code serves either workload.

Sharing one pool of executors can also take less capacity than running two clusters. Batch and interactive load are often uncorrelated, so the capacity one leaves idle is available to the other, and the pool can be sized for the peak of the combined load rather than for the sum of two separate peaks.

Sharing a cluster this way is still early. Pipelined execution is experimental, and the scheduler does not prioritize one job over another, so on a busy cluster an interactive query can wait for vcores that batch jobs hold, as the limitations of pipelined execution describe. Prioritizing interactive queries over batch jobs in the scheduler is what a shared cluster still needs.

Getting Started

The easiest way to get started is to run one of the standalone or distributed examples. After that, refer to the Getting Started Guide.

Web Terminal User Interface (Web TUI)

Ballista provides a browser-based Web TUI for monitoring a running cluster. It exposes the TUI views for jobs, executors, metrics, and scheduler information directly in a web browser.

When the scheduler HTTP endpoint is available, opening the scheduler address in a browser, for example http://localhost:50050, redirects to a hosted Web TUI.

For more information, including how to run the Web TUI locally, see the Ballista CLI documentation.

Project Status

Ballista is part of the Apache DataFusion project at the Apache Software Foundation. It ships about every two months alongside DataFusion. Adopters include Spice AI and Coralogix.

Changes are tested on real clusters before they merge:

• The merge queue runs TPC-H (SF10) and TPC-DS (SF1) queries on a multi-process cluster and compares the results with single-process DataFusion, including runs with multi-partition tasks and, for TPC-H, pipelined mode.
• A fault-injection suite kills executors mid-query, with AQE on and off, and checks that the results are still correct. It runs on every code change, and its Kubernetes scenarios also run nightly.
• Upgrade guides cover breaking changes between releases, and clients get a clear error when their major version doesn‘t match the scheduler’s.

Ballista supports a wide range of SQL, including CTEs, joins and subqueries, and can execute complex queries at scale, but there is still a gap between DataFusion and Ballista, which may bring incompatibilities. The community is actively working to close it. Refer to the DataFusion SQL Reference for more information on supported SQL.

Cargo Features

Ballista uses Cargo features to enable optional functionality.

ballista (client)

ballista-core

ballista-scheduler

ballista-executor

ballista-cli

Usage Examples

# Build with standalone support (default)
cargo build -p ballista

# Build with Substrait support
cargo build -p ballista-scheduler --features substrait

Contribution Guide

Please see the Contribution Guide for information about contributing to Ballista.