import { PartitionSkew } from '@site/src/components/diagrams/PartitionSkew';

# Partitioning 192,000 files

:::warning 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](/glossary/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

```python
@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](/glossary/head-node).

**File sizes vary by 7x.** With one file per task, the largest files become the
[long tail](/lesson/distributed-python-with-eugo-hpc/parallel-fan-out#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

<PartitionSkew />

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:

```python
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.

```python
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

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

Each task returns paths. The resampled rasters go to [object storage](/glossary/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:

```python
# 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](/glossary/long-tail) waiting to
happen.

---

Source: https://university.eugo.io/lesson/scaling-a-real-workload/partitioning