Distributed Evaluation

Scale evaluations across multiple nodes using SLURM, Kubernetes, Ray, or manual sharding.

How Sharding Works

Each worker runs the same nel eval run command. Two environment variables control the split:

nel eval run auto-detects these and evaluates only its assigned problem range.

SLURM

SLURM config file

Use cluster.type: slurm with shards: to enable array job sharding:

1
# slurm_gsm8k_sharded.yaml
2
services:
3
  model:
4
    type: vllm
5
    model: nvidia/Llama-3.1-70B-Instruct
6
    protocol: chat_completions
7
    tensor_parallel_size: 4
8
    port: 8000
9
    node_pool: compute
10
11
benchmarks:
12
  - name: gsm8k
13
    solver:
14
      type: simple
15
      service: model
16
17
cluster:
18
  type: slurm
19
  walltime: "02:00:00"
20
  shards: 16
21
  node_pools:
22
    compute:
23
      partition: batch
24
      nodes: 1
25
      ntasks_per_node: 1
26
      gres: "gpu:4"

Run

$
nel eval run slurm_gsm8k_sharded.yaml

This:

1. Generates eval.sbatch with #SBATCH --array=0-15
2. Each array task exports NEL_SHARD_IDX and NEL_TOTAL_SHARDS
3. Each task writes results to shard_N/ subdirectories

Merge results

After all array tasks complete:

$
nel eval merge ./eval_results

This discovers all shard_N/ directories, deduplicates any overlapping results, and produces merged bundles with combined scores.

Kubernetes

Use a K8s Indexed Job for native parallelism:

$
kubectl apply -f deploy/k8s/eval-indexed-job.yaml

The manifest uses completionMode: Indexed with 8 completions. Each pod gets JOB_COMPLETION_INDEX mapped to NEL_SHARD_IDX:

1
env:
2
  - name: NEL_SHARD_IDX
3
    valueFrom:
4
      fieldRef:
5
        fieldPath: metadata.annotations['batch.kubernetes.io/job-completion-index']
6
  - name: NEL_TOTAL_SHARDS
7
    value: "8"

After all pods complete, merge with nel eval merge.

Ray

For Ray clusters (compatible with NeMo Gym’s Ray infrastructure):

$
ray job submit --working-dir . -- python -m nemo_evaluator.engine.ray_launcher \
>
    --bench gsm8k --shards 8 --repeats 5 \
>
    --output-dir ./eval_results/ray

Each shard runs as a @ray.remote task. Results are merged in-process after all tasks complete.

From Python

1
import ray
2
from nemo_evaluator.engine.ray_launcher import run_shard
3
4
ray.init()
5
futures = [run_shard.remote("gsm8k", i, 8, ...) for i in range(8)]
6
results = ray.get(futures)

Docker Compose

For local multi-container sharding:

$
# 4 shards
$
for i in 0 1 2 3; do
$
  NEL_SHARD_IDX=$i NEL_TOTAL_SHARDS=4 \
>
    docker compose -f deploy/docker-compose.yaml --profile sharded run -d eval-shard
$
done

Manual Sharding (Any Orchestrator)

Works anywhere you can set environment variables:

$
# Terminal 1
$
NEL_SHARD_IDX=0 NEL_TOTAL_SHARDS=4 nel eval run config.yaml -o ./results/shard_0
$
$
# Terminal 2
$
NEL_SHARD_IDX=1 NEL_TOTAL_SHARDS=4 nel eval run config.yaml -o ./results/shard_1
$
$
# ... after all shards complete:
$
nel eval merge ./results

Shard Size Distribution

Problems are distributed evenly. For 14,000 problems across 16 shards:

When the total doesn’t divide evenly, early shards get one extra problem (e.g., 14,001 problems / 16 shards = shard 0 gets 876, shards 1-15 get 875).

Limitations

• shards is incompatible with heterogeneous SLURM jobs (multiple node pools). Use a single pool when sharding.
• shards and auto_resume cannot be used together. Use SLURM --requeue for per-task retries in array mode.
• run_batch() environments (e.g., legacy containers) are not shardable — a warning is emitted if shard env vars are detected.