Spice.ai
DataFusion · Ballista

Building an HA,
Distributed
Query Engine

on Apache DataFusion Ballista

Phillip LeBlanc · Spice AI
00 · About
02 / 14

Phillip LeBlanc

@phillipleblanc

Career

Ecosystem work

DataFusion
01 · Context
03 / 14
Spice 2.0: Real-Time Analytical Query on Operational Data, Without ETL

spice.ai/blog/spice-2-0-is-now-available

02 · Problem
04 / 14

Single-node ceilings.

Scale query compute independently of storage—without Spark-class operational tax, and without giving up a simple single-node deploy.

Capacity

100GB+ lakes
outgrow one process.

Memory and compute hit a hard wall. Analytical work needs stage boundaries, placement, and elastic executors.

×
Fault tolerance

Stages fail.
Processes fail too.

Resume stages from intermediate data. Losing the process that accepted the query should be invisible to the caller—accepted work keeps running elsewhere.

03 · Framework
05 / 14

Why Apache Ballista?

Selected

Apache Ballista

Batteries-included scheduler + executorsDisk-based shuffle · stage retryNative DataFusion · Arrow FlightBet on the Apache community
Compared

datafusion-distributed

Lean, in-memory toolboxFailures restart from scratchDiscovery & cluster semantics left to us
Compared

Custom build

Full control of every layer6–12+ months to production qualityTime better spent on Spice features
Community bet.ASF governance, shared DataFusion ecosystem—we invest upstream instead of owning a private distributed stack.
04 · Extensions
06 / 14

What we built on Ballista.

Stock Ballista is the execution foundation. Production Spice needed control, catalog, security, shuffle choices, and codecs.

01

Poll scheduling

Executor poll loop · no slot leasing · PollNow for latency

02

Catalog + UDFs

Scheduler is source of truth · one spicepod for the cluster

03

mTLS by default

Privileged RPCs on an internal port · CLI cert tooling

04

Shuffle backends

Local disk · memory · object store for ephemeral nodes

05

Vortex shuffle

Smaller intermediates · less network and disk I/O

06

Custom codecs

Serialize Spice plans · UDTFs · Cayenne · Iceberg scans

05 · High availability
07 / 14

Multi-active HA. Object store is all you need.

Every scheduler is active. Shared state lives in the object store—no etcd, ZooKeeper, or Redis.

Scheduler Aown J₁
Scheduler Bown J₂
Scheduler Crecover
ETag conditional writes
Object storecluster.json · topology · heartbeats · job metadata
ANY SCHEDULERClients can hit any node
NO CONSENSUS SERVICEObject store is the only extra dependency
SURVIVES OWNER LOSSAccepted async jobs keep running
06 · Scheduling
08 / 14

Poll for work. Don't lease slots.

Ballista supports both. We pick poll: executors self-regulate—no shared live workload view across schedulers.

Push scheduling

Scheduler pushes tasks

Push needs an accurate shared view of free slots—or you oversubscribe. Slot leasing and coordination become the tax.

  • Scheduler assigns work onto executors
  • Needs a shared slot / workload view
  • Multi-active → lease, lock, or oversubscribe
Poll scheduling · our choice

Executors ask for work

Each executor knows its own capacity. It polls when free—no leasing, no shared workload ledger across schedulers.

  • Executor-initiated poll loop to every scheduler
  • Self-regulates concurrency locally
  • Fits multi-active without slot leasing
PollNowControl-stream wake so poll stays low-latency when work appears—without abandoning self-regulation.
07 · Security
09 / 14

Secure by default. One spicepod.

Privileged cluster RPCs require mTLS. Catalog and UDFs sync from the scheduler—no manual registration drift.

mTLS · internal port 50052

Protect privileged RPCs

50051 / 8090Public query APIsoptional TLS/mTLS
50052SchedulerGrpc · ClusterServicemTLS required
9090Prometheusmetrics
GetAppDefinitionfull cluster config
ExpandSecretsecret values
GetSchedulersscheduler membership / topology
Catalog + UDF sync

Scheduler is source of truth

Executors join and receive catalog + UDF registry via GetCatalog / GetFunctions.

One spicepod.yamlDefines datasets, models, and views for the whole cluster.
SQL todayScheduler plans; stubs ready for future DataFrame / Substrait clients.
No driftNo hand-copied schemas on every client or executor.
08 · Data plane
10 / 14

Shuffle is a deployment choice.

Disk, memory, or object store—plus optional Vortex encoding for smaller intermediates.

Apache Ballista
SSD

Local disk

Default durable path: materialize stage output on the executor.

Stage retry yes*Peer Flight yesNode loss limitedLatency low
Best · stable / colocated nodes
Spice fork
RAM

Memory

Keep intermediates in RAM when recomputation is acceptable.

Stage retry while alivePeer Flight yesNode loss noLatency lowest
Best · short / ephemeral stages
Spice fork
OBJ

Object store

Decouple shuffle from executor disks—essential for ephemeral cloud nodes.

Stage retry yesPeer Flight noNode loss yesLatency highest
Best · elastic / spot executors
VORTEXshuffle_format: vortex — adaptive per-column encoding, smaller shuffle files, less network and disk I/O. Transparent to query semantics.
09 · Demo
11 / 14

DEMO

Scheduler failover for an accepted async query.

10 · Open source
12 / 14

General fixes flow back.

We maintain spiceai/datafusion-ballista—and upstream TLS hooks, shuffle backends, and scheduling fixes to Apache.

Cluster API + hooks4 merged fork PRs

#3#4#11#19

Scheduling + HA14 merged fork PRs

#2#5#6#12#13#14#15#23#26#30#53#56#59#60

Shuffle + transport16 merged fork PRs

#7#8#9#17#18#24#36#40#41#42#43#48#54#61#62#63

Planning + correctness8 merged fork PRs

#1#27#28#29#33#34#57#58

Observability4 merged fork PRs

#10#38#39#49

Maintenance4 merged fork PRs

#20#21#25#47

Apache path

Apache Ballista

Status snapshot · 2026-07-23

direct→#1332 · Unwritable work-dir fixsubmitted 2025-10-23 · landed
#3→#1400 · TLS + custom headerssubmitted 2026-01-23 · landed
#23→#1523 · Lost-executor slot fixsubmitted 2026-03-25 · landed
#14→#1892 · External task semaphoresubmitted 2026-06-23 · landed
#12/#15→#1893/#1891 · PollNow + job eventssubmitted 2026-06-23 · in review
#61→#1951 · Shuffle in-flight governorsubmitted 2026-07-05 · landed · overlap
#59/#60→#2037 · Persist terminal job statesubmitted 2026-07-14 · landed
STATUSlandedin reviewoverlappending
11 · Try it
13 / 14

Get started.

Install Spice, start cluster roles, then follow the async-queries cookbook.

Commands
INSTALLcurl https://install.spiceai.org | /bin/bash
SCHEDULERspiced --role scheduler
EXECUTORspiced --role executor
--scheduler-address https://host:50052
Cookbook

Async queries walkthrough

TLS, scheduler + executor setup, submit/poll, and spice query.

github.com/spiceai/cookbook/tree/trunk/async-queriesOpen the recipe →
Spice.ai
Distributed query without the operational tax.

Ballista for execution.
Object store for HA.

Multi-active schedulers, mTLS, deployable shuffle, and an async API—on the same binary that still runs standalone.

NO ETCD / ZK / REDISSAME SQL · SAME SPICEPODUPSTREAM THE GENERICS
Questions · spice.ai/blog/apache-ballista-at-spice-ai · phillip@spice.ai
Appendix
A1

Executors dial out. Schedulers never dial in.

Control streams are executor-initiated. Object store is the shared plane both sides can reach—including final results.

Reachable · scheduler network
Clients and executors can reach these addresses
Scheduler Ainbound streams
Scheduler Binbound streams
Scheduler Cinbound streams
No scheduler → executor path
Private · executor network
Executor E₇outbound control streams · upload final results to object store
Shared · reachable

Object store

Both schedulers and executors can reach it.
Schedulersmembership · job metadata · recovery graphs
Executorsshuffle (optional) · final result chunks
Spice differenceStock Ballista: clients fetch finals from executors. Spice: executors upload the final result set to object store.

One stream, two directions

Each executor keeps a long-lived outbound stream to every scheduler—commands ride back on that same connection; heartbeats and slot state go up.

PollNowWake the pull loop—cut wait from the poll interval to near-zero.
CancelTasksCancellation rides the open stream—no scheduler dial-out.
MetricsOn-demand executor metrics over the same stream.
HeartbeatSlot availability · liveness from executor to every scheduler.
RESULTS · executor uploads finals to object store · clients need not reach executors
Appendix
A2

Distribute or accelerate.

TPC-H SF100 against federated S3 Parquet—no local acceleration—one node vs a three-executor Ballista cluster.

ConfigurationDurationvs 1-nodeBest fit
Single node · federated S32,741 s · 45.7 min—Baseline without local acceleration
3-executor Ballista cluster935 s · 15.6 min2.9×Distribution when data doesn’t fit locally
Single node · Cayenne accel88 s31.1×Acceleration for working sets that fit
Distribute or accelerateBallista for lake-scale compute · Cayenne when the working set fits.
Spice.aiPRESENTER VIEW
00:00
CURRENT SLIDE
NEXT—