Partitioning 192,000 files

Illustrative figures

The 7 TB / 192,000-file / ~3-minute result is Eugo's published benchmark. The per-file timings, per-stage shares, and scaling numbers in this lesson are illustrative: they show the shape of the finding rather than measurements from that specific run, and are being confirmed against the original profiling data.

The reasoning is the lesson. Measure your own workload for numbers you will act on.

192,000 independent files looks like a solved partitioning problem: one file, one task.

It is not, and the reason is the 7x spread in read times we found while profiling.

The naive version

@eugo.hpc.distribute
def process_file(path):
    raster = read_from_object_storage(path)
    result = resample(decode(raster), TARGET_GRID)
    out = target_path(path)
    write_to_object_storage(out, encode(result))
    return {"path": out, "seconds": ...}

futures = [process_file(p) for p in paths]      # 192,000 tasks
results = eugo.hpc.get(futures)

Correct, and it works. Two problems.

192,000 tasks is a lot of scheduling. At ~0.9 s each, per-task overhead is not fatal, but it is not free either. Dispatching and tracking 192,000 tasks is real work for the head node.

File sizes vary by 7x. With one file per task, the largest files become the long tail: the run finishes when the biggest file finishes, while most workers sit idle waiting for it.

Partition by cost, not by count

Both schedules do identical total work. Dashed outlines are idle workers. Partitioning by file count puts one oversized file alone on worker 2, and adding more workers cannot help, because that partition is already by itself. Partitioning by total bytes finishes in 4 units instead of 7, at the same worker count.

File size is a good proxy for processing cost here. So group files into batches of roughly equal total bytes rather than equal file count:

def batches_by_weight(entries, target_bytes):
    """entries: (path, size_bytes). Yields lists of paths of ~equal total size."""
    batch, weight = [], 0
    for path, size in entries:
        batch.append(path)
        weight += size
        if weight >= target_bytes:
            yield batch
            batch, weight = [], 0
    if batch:
        yield batch

@eugo.hpc.distribute
def process_batch(paths):
    out = []
    for p in paths:
        raster = read_from_object_storage(p)
        result = resample(decode(raster), TARGET_GRID)
        target = target_path(p)
        write_to_object_storage(target, encode(result))
        out.append(target)
    return out                      # small: paths, not rasters

Sorting largest-first before batching helps further: big files get distributed early rather than clustering at the end.

entries = sorted(list_with_sizes(prefix), key=lambda e: -e[1])
batches = list(batches_by_weight(entries, target_bytes=8 * 1024**3))

Choosing the batch size

We targeted ~8 GB per batch, giving roughly 900 batches.

The reasoning:

• Meaningfully more batches than workers. 900 batches across 16 workers is ~56 each, enough for the scheduler to balance uneven batches.
• Task bodies of roughly three minutes. Far above scheduling overhead, and short enough that one slow batch cannot set the runtime for the whole job.
• Bounded memory. A worker holds one file at a time, not a whole batch.

What not to return

# Wrong: gigabytes crossing the network per task
return [resampled_array for ...]

Each task returns paths. The resampled rasters go to object storage from the worker that produced them. Returning the arrays themselves would move the entire 7 TB back through the head node, turning a distributed pipeline into an expensive way to funnel data through one machine.

Return summaries or paths. Never payloads.

Read inside the task

Listing paths in the session is fine, since it is only metadata. Reading content there is not:

# Wrong: the session becomes the bottleneck and its memory the ceiling
rasters = [read_from_object_storage(p) for p in paths]
futures = [process(r) for r in rasters]

# Right: each worker reads its own share
futures = [process_batch(b) for b in batches]

The general rule

Partition by	When
Count	Units are uniform in cost
Size or weight	Units vary, usually the case with files
Grouping key	Downstream aggregation needs keys co-located
Time range	Data is naturally temporal and queries are too

For this workload, weight. The 7x size spread made count-based partitioning a long tail waiting to happen.