Distributed Training & Cloud Planner
Ray-style scatter/gather primitives backed by rayon + auto-scaling planner for datasets >1M rows. Zero infra, pure Python. / Primitives scatter/gather style Ray via rayon + planificateur auto-scaling pour jeux de données >1M lignes.
✓ WorkerPool
✓ cloud_plan
Max workers
auto
0 = rayon thread count
Scatter / gather
✓
allreduce_mean / sum
Planner
✓
in_memory / chunked / streamed
WorkerPool
Create a pool, scatter rows into shards, train per shard in parallel, then allreduce results back.
cloud_plan
Given n_rows × n_cols and a memory budget, produces the optimal strategy: in_memory, chunked or streamed.
cloud_resources
Runtime snapshot: CPU threads, active backend, OS, architecture, registry path.
cloud_count_rows
Streaming CSV row count — constant memory regardless of file size. Useful before calling cloud_plan.
Quick start — WorkerPool
import seraplot as sp, numpy as np
n_rows = 10_000
X = np.random.randn(n_rows, 5)
y = X[:, 0] * 2 + np.random.randn(n_rows) * 0.1
wp = sp.WorkerPool(n_workers=0)
print(f"Pool: {wp.n_workers} workers")
handle = wp.scatter(n_rows)
shards = wp.shards(handle)
coefs = []
for shard in shards:
Xi = X[shard["start"]:shard["end"]]
yi = y[shard["start"]:shard["end"]]
model = sp.LinearRegression()
model.fit(Xi, yi)
coefs.append(model.coef_)
mean_coef = wp.allreduce_mean(coefs)
print(f"Aggregated coef: {mean_coef}")
wp.release(handle)
Quick start — Cloud planner
import seraplot as sp
plan = sp.cloud_plan(n_rows=5_000_000, n_cols=50, mem_budget_mb=4096)
print(f"Strategy : {plan['strategy']}")
print(f"Workers : {plan['recommended_workers']}")
print(f"Chunk rows : {plan['recommended_chunk_rows']}")
print(f"Chunks : {plan['n_chunks']}")
print(f"Est. seconds: {plan['estimated_seconds']:.1f}s")
resources = sp.cloud_resources()
print(f"\nCPU threads : {resources['cpu_threads']}")
print(f"Backend : {resources['backend']}")
print(f"OS : {resources['os']} / {resources['arch']}")
API Reference
WorkerPool
wp = sp.WorkerPool(n_workers=0)
| Method / Attr | Signature | Returns | Description |
|---|---|---|---|
n_workers | property | int | Actual number of workers (resolved from n_workers=0 = rayon thread count) |
scatter | (n_rows) -> int | handle | Partition n_rows rows into shards, return opaque handle |
shards | (handle) -> list[dict] | list[dict] | Get shard list for handle: [{"id":0,"start":0,"end":625}, …] |
release | (handle) | None | Free memory associated with the scatter handle |
allreduce_mean | (vecs: list[list[float]]) -> list[float] | list[float] | Element-wise mean across all vectors (same length) |
allreduce_sum | (vecs: list[list[float]]) -> list[float] | list[float] | Element-wise sum across all vectors |
cloud_plan
| Parameter | Type | Default | Description |
|---|---|---|---|
n_rows | int | — | Number of dataset rows |
n_cols | int | — | Number of feature columns |
mem_budget_mb | int | 2048 | Available RAM budget in MB |
Returns →
dict with keys: n_rows, n_cols, bytes_total, mem_budget_mb, recommended_workers, recommended_chunk_rows, n_chunks, estimated_seconds, strategycloud_resources
Returns →
dict — cpu_threads (int), backend (str), os (str), arch (str), registry_dir (str), tasks_dir (str)cloud_count_rows
| Parameter | Type | Default | Description |
|---|---|---|---|
path | str | — | Path to CSV file |
chunk_rows | int | 100000 | Internal read buffer size |
has_header | bool | True | Skip first line as header |
delimiter | str | "," | CSV delimiter |
Returns →
int — number of data rows (header excluded if has_header=True)Strategy reference (cloud_plan)
| Strategy | Condition | Description |
|---|---|---|
in_memory | Dataset fits in budget | Load all at once, use full parallel training |
chunked | Dataset > budget, <2× budget per chunk | Process in chunks, aggregate results per chunk |
streamed | Dataset >> budget | Stream rows one chunk at a time, online aggregation |
Call
cloud_count_rows(path) on your CSV file, then pass the result to cloud_plan to get the optimal strategy before loading any data into memory.Référence API
WorkerPool
wp = sp.WorkerPool(n_workers=0)
| Méthode / Attr | Signature | Retourne | Description |
|---|---|---|---|
n_workers | propriété | int | Nombre réel de workers (0 = nombre de threads rayon) |
scatter | (n_rows) -> int | handle | Partitionne n_rows lignes en shards, retourne un handle opaque |
shards | (handle) -> list[dict] | list[dict] | Récupère les shards : [{"id":0,"start":0,"end":625}, …] |
release | (handle) | None | Libère la mémoire associée au handle scatter |
allreduce_mean | (vecs) | list[float] | Moyenne élément par élément sur tous les vecteurs |
allreduce_sum | (vecs) | list[float] | Somme élément par élément sur tous les vecteurs |
cloud_plan
| Paramètre | Type | Défaut | Description |
|---|---|---|---|
n_rows | int | — | Nombre de lignes du jeu de données |
n_cols | int | — | Nombre de colonnes de features |
mem_budget_mb | int | 2048 | Budget RAM disponible en Mo |
Retourne →
dict avec les clés : n_rows, n_cols, bytes_total, mem_budget_mb, recommended_workers, recommended_chunk_rows, n_chunks, estimated_seconds, strategycloud_resources
Retourne →
dict — cpu_threads (int), backend (str), os (str), arch (str), registry_dir (str), tasks_dir (str)cloud_count_rows
| Paramètre | Type | Défaut | Description |
|---|---|---|---|
path | str | — | Chemin vers le fichier CSV |
chunk_rows | int | 100000 | Taille du buffer de lecture interne |
has_header | bool | True | Ignorer la première ligne (en-tête) |
delimiter | str | "," | Séparateur CSV |
Retourne →
int — nombre de lignes de données (en-tête exclu si has_header=True)Référence des stratégies (cloud_plan)
| Stratégie | Condition | Description |
|---|---|---|
in_memory | Dataset tient dans le budget | Chargement complet, entraînement parallèle total |
chunked | Dataset > budget, <2× budget par chunk | Traitement en chunks, agrégation des résultats |
streamed | Dataset >> budget | Streaming ligne par ligne, agrégation en ligne |
Appelez
cloud_count_rows(chemin) sur votre CSV, puis passez le résultat à cloud_plan pour obtenir la stratégie optimale avant de charger la moindre donnée en mémoire.