Flink Sort‑Shuffle: Design, Implementation, and Performance Evaluation

Flink is a unified batch‑and‑stream data processing engine, and its ability to handle massive batch jobs depends heavily on the efficiency of the data shuffle phase; the newly introduced sort‑shuffle aims to make this phase more robust and performant.

Traditional batch shuffle implementations use either a hash‑based or a sort‑based model. The hash‑based approach creates many small files, leading to high file‑system metadata overhead, random I/O, and stability problems such as file‑handle exhaustion.

Sort‑shuffle addresses these issues by reducing the number of files, opening fewer files concurrently, maximizing sequential reads/writes, minimizing I/O amplification, and decoupling memory‑buffer consumption from job parallelism, thereby improving both stability and performance.

The design goals are:

Reduce file quantity by writing sorted data to a single file per task.

Open as few files as possible, sharing file handles among downstream readers.

Maximize sequential I/O through larger write batches and an elevator‑style I/O scheduler.

Avoid I/O amplification by eliminating the merge step.

Limit memory‑buffer usage regardless of parallelism.

Implementation details include a memory sort buffer defined by the following interface:

public interface SortBuffer {
   /** Appends data of the specified channel to this SortBuffer. */
   boolean append(ByteBuffer source, int targetChannel, Buffer.DataType dataType) throws IOException;

   /** Copies data in this SortBuffer to the target MemorySegment. */
   BufferWithChannel copyIntoSegment(MemorySegment target);

   long numRecords();
   long numBytes();
   boolean hasRemaining();
   void finish();
   boolean isFinished();
   void release();
   boolean isReleased();
}

The system uses a bucket‑sort algorithm where each serialized record carries a 16‑byte metadata header (length, type, and a pointer to the next record in the same partition). Files contain multiple data regions, each representing a sort‑spill, and an accompanying index file maps partitions to offsets and lengths.

The I/O scheduler follows an elevator‑like algorithm, illustrated by the following pseudo‑code:

for (data_region in data_regions) {
    data_reader = poll_reader_of_the_smallest_file_offset(data_readers);
    if (data_reader == null) break;
    reading_buffers = request_reading_buffers();
    if (reading_buffers.isEmpty()) break;
    read_data(data_region, data_reader, reading_buffers);
}

Additional optimizations include broadcast data deduplication—storing a single copy of broadcast records in both memory buffers and shuffle files—and optional compression, which can improve TPC‑DS performance by over 30%.

Test results show that the sort‑shuffle dramatically improves stability (eliminating file‑handle and inode exhaustion issues) and delivers 2‑6× overall speedup for 10 TB TPC‑DS workloads, with up to 10× improvement when measuring shuffle time alone.

Configuration tips: enable sort‑shuffle by setting taskmanager.network.sort-shuffle.min-parallelism to 1 on HDDs, turn on compression via taskmanager.network.blocking-shuffle.compression.enabled , and tune buffer counts with taskmanager.network.sort-shuffle.min-buffers to avoid OOM.

Future work includes network connection reuse, multi‑disk load balancing, a remote shuffle service, and user‑selectable disk types (HDD vs. SSD).