You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by gi...@apache.org on 2026/04/27 21:58:20 UTC
(datafusion-ballista) branch asf-site updated: Publish built docs triggered by df79b8408012bba5607158716715e7e10f12416f
This is an automated email from the ASF dual-hosted git repository.
github-actions[bot] pushed a commit to branch asf-site
in repository https://gitbox.apache.org/repos/asf/datafusion-ballista.git
The following commit(s) were added to refs/heads/asf-site by this push:
new ebadd4ac6 Publish built docs triggered by df79b8408012bba5607158716715e7e10f12416f
ebadd4ac6 is described below
commit ebadd4ac63b6cd78004b818c59b7c8b0b93a0919
Author: github-actions[bot] <gi...@users.noreply.github.com>
AuthorDate: Mon Apr 27 21:58:16 2026 +0000
Publish built docs triggered by df79b8408012bba5607158716715e7e10f12416f
---
_sources/user-guide/configs.md.txt | 17 ++++++
_sources/user-guide/tuning-guide.md.txt | 51 ++++++++++++++++++
searchindex.js | 2 +-
user-guide/configs.html | 62 ++++++++++++++++++++++
user-guide/extensions-example.html | 2 +-
user-guide/tuning-guide.html | 91 +++++++++++++++++++++++++++++++++
6 files changed, 223 insertions(+), 2 deletions(-)
diff --git a/_sources/user-guide/configs.md.txt b/_sources/user-guide/configs.md.txt
index dbc996eb4..cd0a790ff 100644
--- a/_sources/user-guide/configs.md.txt
+++ b/_sources/user-guide/configs.md.txt
@@ -88,6 +88,23 @@ let expected = [
];
```
+## Shuffle Settings
+
+The following session-level keys control Ballista's shuffle behavior. See
+the [tuning guide](tuning-guide.md#shuffle-implementation) for an
+explanation of the hash-based (default) and sort-based shuffle writers.
+
+| key | type | default | description |
+| --------------------------------------------- | ------- | --------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
+| ballista.shuffle.max_concurrent_read_requests | UInt64 | 64 | Maximum number of concurrent fetch requests the shuffle reader will issue. |
+| ballista.shuffle.force_remote_read | Boolean | false | Forces the shuffle reader to fetch every partition through Arrow Flight, even when the data is local. Intended for testing. |
+| ballista.shuffle.remote_read_prefer_flight | Boolean | false | For remote reads, prefer the Arrow Flight reader over the block reader. The block reader is generally faster. |
+| ballista.shuffle.sort_based.enabled | Boolean | false | Enables the sort-based shuffle writer (consolidated data file per input partition with an index, instead of one file per (input partition, output partition) pair). |
+| ballista.shuffle.sort_based.buffer_size | UInt64 | 1048576 | Per-partition buffer size in bytes for the sort-based writer (1 MiB default). |
+| ballista.shuffle.sort_based.memory_limit | UInt64 | 268435456 | Total in-memory budget across all output-partition buffers for the sort-based writer (256 MiB default). |
+| ballista.shuffle.sort_based.spill_threshold | Utf8 | "0.8" | Fraction of `memory_limit` at which the largest buffers spill to disk. Must be in the range 0–1. |
+| ballista.shuffle.sort_based.batch_size | UInt64 | 8192 | Target row count when coalescing buffered batches before they are written or spilled. |
+
## Ballista Scheduler Configuration Settings
Besides the BallistaContext configuration settings, a few configuration settings for the Ballista scheduler to better
diff --git a/_sources/user-guide/tuning-guide.md.txt b/_sources/user-guide/tuning-guide.md.txt
index fe4363df3..dfe3c2937 100644
--- a/_sources/user-guide/tuning-guide.md.txt
+++ b/_sources/user-guide/tuning-guide.md.txt
@@ -66,6 +66,57 @@ decreasing the number of concurrent tasks may help.
In the future, Ballista will have better support for tracking memory usage and allocating tasks based on available
memory, as well as supporting spill-to-disk to reduce memory pressure.
+## Shuffle Implementation
+
+Ballista exchanges data between query stages by writing the output of each
+upstream task to local files, which downstream tasks read either from disk
+(when co-located) or over Arrow Flight. Two shuffle implementations are
+available, with different trade-offs around file count, memory use, and
+write latency.
+
+### Hash-based shuffle (default)
+
+The default writer hashes each incoming `RecordBatch` and immediately
+encodes the per-partition slices to Arrow IPC, streaming them into one
+file per `(input_partition, output_partition)` pair. Nothing is buffered
+in memory across batches.
+
+This is simple and low latency, but for `N` input partitions and `M`
+output partitions it produces `N × M` files. Wide shuffles can therefore
+generate a large number of small files.
+
+### Sort-based shuffle (opt-in)
+
+The sort-based writer accumulates batches in a per-output-partition
+in-memory buffer. When the total buffered size crosses a threshold, the
+largest buffers are spilled to disk. After the input stream finishes, the
+remaining in-memory data and any spilled batches are merged and written
+into a single consolidated Arrow IPC file per input partition, alongside
+an index file that lets readers seek directly to a given output partition.
+
+This produces `2 × N` files instead of `N × M`, coalesces small batches
+to a target size before writing, and bounds shuffle memory use via
+spilling — at the cost of higher write latency than the hash writer.
+Consider enabling it for queries with high partition fan-out or when
+file-count pressure on local storage is a concern.
+
+The sort-based writer is disabled by default and is enabled per session:
+
+```rust
+let session_config = SessionConfig::new_with_ballista()
+ .set_bool("ballista.shuffle.sort_based.enabled", true);
+```
+
+The following session-level keys tune its behavior:
+
+| key | type | default | description |
+| ------------------------------------------- | ------- | --------- | --------------------------------------------------------------------------------------------------------- |
+| ballista.shuffle.sort_based.enabled | Boolean | false | Enables the sort-based shuffle writer. |
+| ballista.shuffle.sort_based.buffer_size | UInt64 | 1048576 | Per-partition buffer size in bytes (1 MiB default). |
+| ballista.shuffle.sort_based.memory_limit | UInt64 | 268435456 | Total in-memory budget across all output-partition buffers (256 MiB default). |
+| ballista.shuffle.sort_based.spill_threshold | Utf8 | "0.8" | Fraction of `memory_limit` at which the largest buffers begin spilling to disk. Must be in the range 0–1. |
+| ballista.shuffle.sort_based.batch_size | UInt64 | 8192 | Target row count when coalescing buffered batches before they are written or spilled. |
+
## Push-based vs Pull-based Task Scheduling
Ballista supports both push-based and pull-based task scheduling. It is recommended that you try both to determine
diff --git a/searchindex.js b/searchindex.js
index 8a6ff0507..5ed571dda 100644
--- a/searchindex.js
+++ b/searchindex.js
@@ -1 +1 @@
-Search.setIndex({"alltitles": {"Aggregate Functions": [[21, "aggregate-functions"]], "Apache DataFusion Ballista": [[4, null]], "Arrow-native": [[1, "arrow-native"]], "Autoscaling Executors": [[11, "autoscaling-executors"]], "Ballista Architecture": [[1, null]], "Ballista Code Organization": [[2, null]], "Ballista Command-line Interface": [[5, null]], "Ballista Configuration Settings": [[6, "ballista-configuration-settings"]], "Ballista Development": [[3, null]], "Ballista Python Binding [...]
\ No newline at end of file
+Search.setIndex({"alltitles": {"Aggregate Functions": [[21, "aggregate-functions"]], "Apache DataFusion Ballista": [[4, null]], "Arrow-native": [[1, "arrow-native"]], "Autoscaling Executors": [[11, "autoscaling-executors"]], "Ballista Architecture": [[1, null]], "Ballista Code Organization": [[2, null]], "Ballista Command-line Interface": [[5, null]], "Ballista Configuration Settings": [[6, "ballista-configuration-settings"]], "Ballista Development": [[3, null]], "Ballista Python Binding [...]
\ No newline at end of file
diff --git a/user-guide/configs.html b/user-guide/configs.html
index a6e2bc19d..21c111643 100644
--- a/user-guide/configs.html
+++ b/user-guide/configs.html
@@ -270,6 +270,11 @@
Ballista Configuration Settings
</a>
</li>
+ <li class="toc-h2 nav-item toc-entry">
+ <a class="reference internal nav-link" href="#shuffle-settings">
+ Shuffle Settings
+ </a>
+ </li>
<li class="toc-h2 nav-item toc-entry">
<a class="reference internal nav-link" href="#ballista-scheduler-configuration-settings">
Ballista Scheduler Configuration Settings
@@ -383,6 +388,63 @@
</pre></div>
</div>
</section>
+<section id="shuffle-settings">
+<h2>Shuffle Settings<a class="headerlink" href="#shuffle-settings" title="Link to this heading">¶</a></h2>
+<p>The following session-level keys control Ballista’s shuffle behavior. See
+the <a class="reference internal" href="tuning-guide.html#shuffle-implementation"><span class="std std-ref">tuning guide</span></a> for an
+explanation of the hash-based (default) and sort-based shuffle writers.</p>
+<table class="table">
+<thead>
+<tr class="row-odd"><th class="head"><p>key</p></th>
+<th class="head"><p>type</p></th>
+<th class="head"><p>default</p></th>
+<th class="head"><p>description</p></th>
+</tr>
+</thead>
+<tbody>
+<tr class="row-even"><td><p>ballista.shuffle.max_concurrent_read_requests</p></td>
+<td><p>UInt64</p></td>
+<td><p>64</p></td>
+<td><p>Maximum number of concurrent fetch requests the shuffle reader will issue.</p></td>
+</tr>
+<tr class="row-odd"><td><p>ballista.shuffle.force_remote_read</p></td>
+<td><p>Boolean</p></td>
+<td><p>false</p></td>
+<td><p>Forces the shuffle reader to fetch every partition through Arrow Flight, even when the data is local. Intended for testing.</p></td>
+</tr>
+<tr class="row-even"><td><p>ballista.shuffle.remote_read_prefer_flight</p></td>
+<td><p>Boolean</p></td>
+<td><p>false</p></td>
+<td><p>For remote reads, prefer the Arrow Flight reader over the block reader. The block reader is generally faster.</p></td>
+</tr>
+<tr class="row-odd"><td><p>ballista.shuffle.sort_based.enabled</p></td>
+<td><p>Boolean</p></td>
+<td><p>false</p></td>
+<td><p>Enables the sort-based shuffle writer (consolidated data file per input partition with an index, instead of one file per (input partition, output partition) pair).</p></td>
+</tr>
+<tr class="row-even"><td><p>ballista.shuffle.sort_based.buffer_size</p></td>
+<td><p>UInt64</p></td>
+<td><p>1048576</p></td>
+<td><p>Per-partition buffer size in bytes for the sort-based writer (1 MiB default).</p></td>
+</tr>
+<tr class="row-odd"><td><p>ballista.shuffle.sort_based.memory_limit</p></td>
+<td><p>UInt64</p></td>
+<td><p>268435456</p></td>
+<td><p>Total in-memory budget across all output-partition buffers for the sort-based writer (256 MiB default).</p></td>
+</tr>
+<tr class="row-even"><td><p>ballista.shuffle.sort_based.spill_threshold</p></td>
+<td><p>Utf8</p></td>
+<td><p>“0.8”</p></td>
+<td><p>Fraction of <code class="docutils literal notranslate"><span class="pre">memory_limit</span></code> at which the largest buffers spill to disk. Must be in the range 0–1.</p></td>
+</tr>
+<tr class="row-odd"><td><p>ballista.shuffle.sort_based.batch_size</p></td>
+<td><p>UInt64</p></td>
+<td><p>8192</p></td>
+<td><p>Target row count when coalescing buffered batches before they are written or spilled.</p></td>
+</tr>
+</tbody>
+</table>
+</section>
<section id="ballista-scheduler-configuration-settings">
<h2>Ballista Scheduler Configuration Settings<a class="headerlink" href="#ballista-scheduler-configuration-settings" title="Link to this heading">¶</a></h2>
<p>Besides the BallistaContext configuration settings, a few configuration settings for the Ballista scheduler to better
diff --git a/user-guide/extensions-example.html b/user-guide/extensions-example.html
index 5b6d6f30f..2f117f426 100644
--- a/user-guide/extensions-example.html
+++ b/user-guide/extensions-example.html
@@ -523,7 +523,7 @@ scheduler.</p>
<p>The logical extension codec typically consists of two components: Google Protocol Buffer definitions:</p>
<div class="highlight-proto notranslate"><div class="highlight"><pre><span></span><span class="k">syntax</span><span class="w"> </span><span class="o">=</span><span class="w"> </span><span class="s">"proto3"</span><span class="p">;</span>
-<span class="kn">package</span><span class="w"> </span><span class="nn">extension</span><span class="o">.</span><span class="n">ballista</span><span class="p">;</span>
+<span class="kn">package</span><span class="w"> </span><span class="nn">extension.ballista</span><span class="p">;</span>
<span class="c1">// used in building the codecs and compiled to Rust at build time.</span>
<span class="k">import</span><span class="w"> </span><span class="s">"datafusion_common.proto"</span><span class="p">;</span>
diff --git a/user-guide/tuning-guide.html b/user-guide/tuning-guide.html
index 428b2b011..13b77c6fe 100644
--- a/user-guide/tuning-guide.html
+++ b/user-guide/tuning-guide.html
@@ -275,6 +275,23 @@
Configuring Executor Concurrency Levels
</a>
</li>
+ <li class="toc-h2 nav-item toc-entry">
+ <a class="reference internal nav-link" href="#shuffle-implementation">
+ Shuffle Implementation
+ </a>
+ <ul class="nav section-nav flex-column">
+ <li class="toc-h3 nav-item toc-entry">
+ <a class="reference internal nav-link" href="#hash-based-shuffle-default">
+ Hash-based shuffle (default)
+ </a>
+ </li>
+ <li class="toc-h3 nav-item toc-entry">
+ <a class="reference internal nav-link" href="#sort-based-shuffle-opt-in">
+ Sort-based shuffle (opt-in)
+ </a>
+ </li>
+ </ul>
+ </li>
<li class="toc-h2 nav-item toc-entry">
<a class="reference internal nav-link" href="#push-based-vs-pull-based-task-scheduling">
Push-based vs Pull-based Task Scheduling
@@ -373,6 +390,80 @@ decreasing the number of concurrent tasks may help.</p>
<p>In the future, Ballista will have better support for tracking memory usage and allocating tasks based on available
memory, as well as supporting spill-to-disk to reduce memory pressure.</p>
</section>
+<section id="shuffle-implementation">
+<h2>Shuffle Implementation<a class="headerlink" href="#shuffle-implementation" title="Link to this heading">¶</a></h2>
+<p>Ballista exchanges data between query stages by writing the output of each
+upstream task to local files, which downstream tasks read either from disk
+(when co-located) or over Arrow Flight. Two shuffle implementations are
+available, with different trade-offs around file count, memory use, and
+write latency.</p>
+<section id="hash-based-shuffle-default">
+<h3>Hash-based shuffle (default)<a class="headerlink" href="#hash-based-shuffle-default" title="Link to this heading">¶</a></h3>
+<p>The default writer hashes each incoming <code class="docutils literal notranslate"><span class="pre">RecordBatch</span></code> and immediately
+encodes the per-partition slices to Arrow IPC, streaming them into one
+file per <code class="docutils literal notranslate"><span class="pre">(input_partition,</span> <span class="pre">output_partition)</span></code> pair. Nothing is buffered
+in memory across batches.</p>
+<p>This is simple and low latency, but for <code class="docutils literal notranslate"><span class="pre">N</span></code> input partitions and <code class="docutils literal notranslate"><span class="pre">M</span></code>
+output partitions it produces <code class="docutils literal notranslate"><span class="pre">N</span> <span class="pre">×</span> <span class="pre">M</span></code> files. Wide shuffles can therefore
+generate a large number of small files.</p>
+</section>
+<section id="sort-based-shuffle-opt-in">
+<h3>Sort-based shuffle (opt-in)<a class="headerlink" href="#sort-based-shuffle-opt-in" title="Link to this heading">¶</a></h3>
+<p>The sort-based writer accumulates batches in a per-output-partition
+in-memory buffer. When the total buffered size crosses a threshold, the
+largest buffers are spilled to disk. After the input stream finishes, the
+remaining in-memory data and any spilled batches are merged and written
+into a single consolidated Arrow IPC file per input partition, alongside
+an index file that lets readers seek directly to a given output partition.</p>
+<p>This produces <code class="docutils literal notranslate"><span class="pre">2</span> <span class="pre">×</span> <span class="pre">N</span></code> files instead of <code class="docutils literal notranslate"><span class="pre">N</span> <span class="pre">×</span> <span class="pre">M</span></code>, coalesces small batches
+to a target size before writing, and bounds shuffle memory use via
+spilling — at the cost of higher write latency than the hash writer.
+Consider enabling it for queries with high partition fan-out or when
+file-count pressure on local storage is a concern.</p>
+<p>The sort-based writer is disabled by default and is enabled per session:</p>
+<div class="highlight-rust notranslate"><div class="highlight"><pre><span></span><span class="kd">let</span><span class="w"> </span><span class="n">session_config</span><span class="w"> </span><span class="o">=</span><span class="w"> </span><span class="n">SessionConfig</span><span class="p">::</span><span class="n">new_with_ballista</span><span class="p">()</span>
+<span class="w"> </span><span class="p">.</span><span class="n">set_bool</span><span class="p">(</span><span class="s">"ballista.shuffle.sort_based.enabled"</span><span class="p">,</span><span class="w"> </span><span class="kc">true</span><span class="p">);</span>
+</pre></div>
+</div>
+<p>The following session-level keys tune its behavior:</p>
+<table class="table">
+<thead>
+<tr class="row-odd"><th class="head"><p>key</p></th>
+<th class="head"><p>type</p></th>
+<th class="head"><p>default</p></th>
+<th class="head"><p>description</p></th>
+</tr>
+</thead>
+<tbody>
+<tr class="row-even"><td><p>ballista.shuffle.sort_based.enabled</p></td>
+<td><p>Boolean</p></td>
+<td><p>false</p></td>
+<td><p>Enables the sort-based shuffle writer.</p></td>
+</tr>
+<tr class="row-odd"><td><p>ballista.shuffle.sort_based.buffer_size</p></td>
+<td><p>UInt64</p></td>
+<td><p>1048576</p></td>
+<td><p>Per-partition buffer size in bytes (1 MiB default).</p></td>
+</tr>
+<tr class="row-even"><td><p>ballista.shuffle.sort_based.memory_limit</p></td>
+<td><p>UInt64</p></td>
+<td><p>268435456</p></td>
+<td><p>Total in-memory budget across all output-partition buffers (256 MiB default).</p></td>
+</tr>
+<tr class="row-odd"><td><p>ballista.shuffle.sort_based.spill_threshold</p></td>
+<td><p>Utf8</p></td>
+<td><p>“0.8”</p></td>
+<td><p>Fraction of <code class="docutils literal notranslate"><span class="pre">memory_limit</span></code> at which the largest buffers begin spilling to disk. Must be in the range 0–1.</p></td>
+</tr>
+<tr class="row-even"><td><p>ballista.shuffle.sort_based.batch_size</p></td>
+<td><p>UInt64</p></td>
+<td><p>8192</p></td>
+<td><p>Target row count when coalescing buffered batches before they are written or spilled.</p></td>
+</tr>
+</tbody>
+</table>
+</section>
+</section>
<section id="push-based-vs-pull-based-task-scheduling">
<h2>Push-based vs Pull-based Task Scheduling<a class="headerlink" href="#push-based-vs-pull-based-task-scheduling" title="Link to this heading">¶</a></h2>
<p>Ballista supports both push-based and pull-based task scheduling. It is recommended that you try both to determine
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org