Input Splits Determine Task Granularity in Batch Jobs

A batch framework decides how a job’s work is spread across machines before it reads a single record. The input format is asked to enumerate splits, each split being a byte range plus the hosts that hold it, and each split becomes one task. File size does not determine the task; split size does. A 10 GiB file in a distributed filesystem becomes on the order of eighty splits under default settings, and the arithmetic that produces that number is the arithmetic that decides how well the cluster is used.

Split derivation from filesystem blocks

A distributed filesystem stores each file as blocks of fixed size, 128 MiB in current releases and 64 MiB in older configurations, and replicates each block to several datanodes. The default input format keeps splits aligned with those boundaries: split size defaults to the block size, and the file is covered by consecutive ranges of that length. A split object is small metadata — a start offset, a length, and a list of hostnames — and the framework consumes the list in the order the directory listing produced it.

Alignment with blocks matters because it makes a split readable from a single replica. A split straddling two blocks would need two network streams and two sets of block checksums, so formats that support splitting avoid it. Reading a split is a seek followed by a sequential read of its length in bytes, so the cost of a split is roughly the cost of streaming those bytes plus the fixed cost of starting a task.

The host list is the basis for locality. The scheduler prefers a node holding a replica, then a node in the same rack, then any node. The three cases are conventionally called node-local, rack-local, and off-switch, and their costs differ by roughly the ratio between local disk throughput, hundreds of megabytes per second, and what a shared rack link delivers. When the cluster is busy, every replica of a split may be occupied, and the job quietly pays network cost for data that was nominally local.

Object stores break this model. They expose a flat namespace of keys with no block layout and no replica affinity, so splits are still byte ranges but no host can be preferred, and each range read is a request whose per-request overhead is large relative to the bytes returned. A dataset of a million small objects is far slower to scan than the same bytes in one object, and the remedy is the same as for small files in a distributed filesystem: fewer, larger units.

The fixed cost of a task

A task is not free. Starting one involves launching a process or container, negotiating resources with the cluster manager, initializing the input reader, and, in JVM-based frameworks, paying class loading and just-in-time warm-up. That fixed cost is measured in hundreds of milliseconds at best and seconds under contention. A task reading a 128 MiB split moves enough bytes to bury the setup cost. A task reading a 1 MiB split spends most of its scheduled time not reading.

The ratio of fixed cost to work sets a practical floor on split size. Splits of tens of megabytes keep overhead in the low single-digit percentage of task runtime; splits of kilobytes keep the cluster busier starting tasks than processing data.

Split count also sets a concurrency ceiling. With one task per split, a job can never use more slots than it has splits, so four splits on a hundred-node cluster leave the cluster idle. The common target is a small multiple of available slots — enough splits that one slow task finishing late does not extend the stage, but not so many that scheduling overhead dominates. Two to four splits per slot is the usual shape of that compromise.

The small-file problem

A file shorter than the split size becomes exactly one split, because a split cannot be cut below the length of the file without wasting readers. A directory of 100,000 files of 1 MiB each therefore produces 100,000 tasks for 100 GiB of data, and each task’s fixed cost exceeds its read time by a wide margin.

Metadata compounds this. A distributed filesystem keeps a block map for every file in the coordinator’s heap, on the order of a hundred bytes of heap per block record in the reference implementation, which turns a namespace of hundreds of millions of small files into gigabytes of memory spent on names rather than data. Directory listings, renames, and recursive scans all degrade as that map grows.

Frameworks answer with combining input formats that pack many small files into one split, so a task reads a few hundred megabytes of small files in sequence. That recovers throughput at the cost of locality: a combined split spans several datanodes, so at most one of its files is read locally and the remainder crosses the network. The alternative is to rewrite the layout upstream into files near the target split size, which is the only option that removes both the task-count problem and the metadata problem.

Splits that cannot be cut at any byte

Most file formats cannot be split at an arbitrary offset, because a record may straddle the boundary. Structured formats solve this with a sync marker: the writer emits a distinctive byte sequence every so many records and records the offsets, and a reader starting at a split boundary seeks forward to the first sync marker rather than trusting the offset. The record that straddles the boundary is then handed to exactly one reader, by the rule that a task skips its first record unless its range starts at offset zero. Without that rule, the trailing record of one split and the leading record of the next would both be counted.

Compression divides the same problem differently. Deflate streams, and therefore the gzip member format built on deflate, are not seekable: decompression state depends on every bit that came before, so a reader must start at the beginning of a member. A gzip file therefore yields exactly one split regardless of size. The consequence is stark — 100 GiB of line-delimited gzip is one task on one core, at any cluster size — which is why compression choice is really a parallelism decision. Concatenating independent gzip members in one file is common, but a general reader cannot seek into the middle of the concatenation without scanning, so the split count stays at one.

bzip2 behaves differently because its format is built from independent blocks holding up to 900 KiB of input each, compressed separately. A reader can seek to a block boundary and decode forward, so a bzip2 stream splits at block granularity; because the blocks are small, frameworks group many of them into one split. Columnar formats expose internal boundaries in the same way — row groups in Parquet, stripes in ORC — and their readers split there rather than on raw bytes.

Split shape is not only a performance detail. It fixes the number of map tasks, which fixes the number of shuffle outputs and the fetch fan-in the reduce side will face. Those numbers show up in the partitioner’s arithmetic and are paid for in the shuffle data path, so split sizing is the first decision in the chain rather than a storage concern.

Boundary cases that change the count

Several cases make the split list differ from what a file listing suggests. A zero-length file produces either no split or an empty one depending on the format, and empty splits still occupy a task slot. The split list is computed once, at submission, from a directory listing taken at that moment, so data arriving while the job runs is invisible and data deleted after listing causes a read failure that fails the task. A file that grows between listing and read yields a split whose recorded length no longer covers the file, and a reader that trusts the length silently ignores the tail.

Compressed input makes the parallelism estimate hard to state in advance. The split count equals the file count rather than the byte count, so the task count depends on a compression ratio that is not known until the data is written. Unsplittable input also interacts badly with skew: when one input file is one task, imbalance among the files becomes imbalance in the schedule, and no partitioner downstream can repair it.