Parallelism — multiprocessing, joblib, and Dask

Why this matters

Once a single core isn't enough, you need real parallelism. CPU-bound work scales with processes (or free-threaded Python); embarrassingly parallel data work with Dask / Ray; distributed work with Spark / Ray cluster. Knowing which to reach for prevents over-engineering (don't deploy Spark for 4 GB) and under-engineering (don't try to fit 200 GB on one machine).

Learning objectives

1. Use multiprocessing and ProcessPoolExecutor cleanly.
2. Use shared memory and mp.Manager correctly.
3. Use Dask for out-of-core / multi-machine arrays and dataframes.
4. Use Ray for distributed actors and parallel tasks.
5. Know when MPI / Spark / Slurm is right.

1. The decision rule (revisit)

Most "I need parallelism" problems stop at ProcessPoolExecutor or Dask. Reach further only when you've measured.

2. multiprocessing recap

(See Lesson 4.3.) Quick reference:

from concurrent.futures import ProcessPoolExecutor

def heavy(x): ...

if __name__ == "__main__":                       # required on Windows / macOS
    with ProcessPoolExecutor(max_workers=8) as ex:
        results = list(ex.map(heavy, items, chunksize=100))

• Each worker has its own memory; no shared state by default.
• Arguments + return values are pickled between processes.
• chunksize matters — too small = pickling dominates.

Shared memory

For huge read-only data (e.g., a NumPy array workers all read):

from multiprocessing import shared_memory
import numpy as np

a = np.ones(10_000_000, dtype=np.float32)
shm = shared_memory.SharedMemory(create=True, size=a.nbytes)
shared_a = np.ndarray(a.shape, dtype=a.dtype, buffer=shm.buf)
shared_a[:] = a

# Worker process:
existing = shared_memory.SharedMemory(name=shm.name)
view = np.ndarray(a.shape, dtype=a.dtype, buffer=existing.buf)
# read freely; do NOT mutate without coordination

Avoids re-pickling GB of data per task. Available since 3.8.

mp.Manager for shared mutable objects

from multiprocessing import Manager
with Manager() as m:
    counter = m.Value("i", 0)
    d = m.dict()
    lock = m.Lock()
    # pass `counter`, `d`, `lock` to workers

Manager proxies are convenient but slow (every access is an RPC). Use only for low-throughput sharing.

3. mpire — multiprocessing without footguns

uv add mpire

from mpire import WorkerPool

with WorkerPool(n_jobs=8) as pool:
    results = pool.map(heavy, items, progress_bar=True)

Wraps multiprocessing with:

• Built-in progress bars.
• Better error reporting.
• Shared objects.
• imap_unordered for streaming.

Drop-in for many tasks where the stdlib is fiddly.

4. Dask — pandas/NumPy at scale

uv add "dask[complete]"

Dask is two things:

1. Bigger-than-RAM versions of NumPy / pandas APIs.
2. A task scheduler that runs anywhere from one machine to thousands.

Dask DataFrame (pandas-like)

import dask.dataframe as dd

df = dd.read_parquet("s3://bucket/data/*.parquet")
result = (df[df.amount > 100]
          .groupby("region")["amount"].sum()
          .compute())                          # actually runs

API mirrors pandas. Operations are lazy until .compute().

Dask Array (NumPy-like)

import dask.array as da
a = da.from_array(huge_numpy_array, chunks=(1000, 1000))
result = (a.mean(axis=0) + a.std(axis=0)).compute()

Computes in chunks; never materialises the whole array.

Dask Delayed (general)

from dask import delayed, compute

@delayed
def step(x): return slow(x)

tasks = [step(x) for x in inputs]
results = compute(*tasks)                       # parallel execution

Wraps any function call as a node in a task graph.

Cluster mode

from dask.distributed import Client
client = Client(n_workers=4, threads_per_worker=2)
# or attach to a remote cluster

Dashboard at http://localhost:8787 shows live tasks, memory, network — incredibly useful.

When use Dask

• Single-machine datasets bigger than RAM.
• Multi-machine clusters managed yourself or via dask-kubernetes / dask-gateway.
• Familiar pandas/NumPy API.

Polars / DuckDB now handle a lot of single-machine "out of core" work more efficiently. Dask wins when you need pandas semantics or genuine multi-machine compute.

5. Ray — distributed tasks + actors

uv add "ray[default]"

import ray
ray.init()                                       # single node; or address="auto" for cluster

@ray.remote
def heavy(x):
    return slow(x)

futures = [heavy.remote(x) for x in items]
results = ray.get(futures)

Resources

@ray.remote(num_cpus=2, num_gpus=0.5)
def train_chunk(data): ...

Schedule by resource — Ray's selling point for ML workloads.

Actors — stateful workers

@ray.remote
class Counter:
    def __init__(self): self.n = 0
    def inc(self): self.n += 1; return self.n

c = Counter.remote()
ray.get(c.inc.remote())                          # 1
ray.get(c.inc.remote())                          # 2

Actors keep state across calls. Perfect for caches, parameter servers, RL environments.

Ray Data

A modern alternative to Spark for data preprocessing in ML pipelines:

import ray.data as rd
ds = rd.read_parquet("s3://bucket/*.parquet").map_batches(preprocess).write_parquet("out/")

Streaming, lazy, scales to clusters. Used heavily by OpenAI / Anthropic for data prep.

When use Ray

• ML training / inference at scale.
• Stateful distributed workloads (RL, parameter servers, agent simulations).
• Need to scale up and down on demand.

6. Apache Spark via PySpark

Spark dominates enterprise big-data jobs. Use when:

• Working with petabytes.
• Existing Spark cluster at your company.
• SQL or DataFrame transformations on huge data.

from pyspark.sql import SparkSession, functions as F

spark = SparkSession.builder.appName("etl").getOrCreate()
df = spark.read.parquet("s3://bucket/data/")
result = (df.filter(df.amount > 100)
            .groupBy("region").agg(F.sum("amount")))
result.write.parquet("out/")

Spark is heavy; for under ~100 GB single-machine work, Polars / DuckDB / Dask usually win.

7. MPI for HPC

uv add mpi4py

from mpi4py import MPI
comm = MPI.COMM_WORLD
rank = comm.Get_rank()
size = comm.Get_size()

if rank == 0:
    data = compute()
    comm.send(data, dest=1, tag=42)
elif rank == 1:
    data = comm.recv(source=0, tag=42)

Run with:

mpiexec -n 4 python script.py

For tightly-coupled HPC (climate sims, molecular dynamics). Rare in mainstream data / AI.

8. Patterns

Map-reduce on a pool

with ProcessPoolExecutor() as ex:
    counts = list(ex.map(count_in_chunk, file_chunks))
total = sum(counts)

Streaming + bounded buffer

from concurrent.futures import ProcessPoolExecutor
import asyncio

async def main():
    loop = asyncio.get_running_loop()
    with ProcessPoolExecutor() as ex:
        async for chunk in stream_chunks():
            result = await loop.run_in_executor(ex, heavy, chunk)
            yield result

GPU/accelerator parallelism

For neural-net training, use PyTorch DDP / FSDP / DeepSpeed / JAX pmap — these are the right abstractions, not Dask/Ray for the GPU layer.

9. Cost / complexity trade-offs

Pick the smallest tool that fits.

10. Worked example: parallel feature engineering

A 30 GB CSV needs per-row transforms then aggregation.

Option A: chunked single-machine

import pandas as pd
def process_chunk(chunk):
    chunk["x2"] = chunk["x"] ** 2
    return chunk.groupby("region")["x2"].sum()

partials = []
for chunk in pd.read_csv("big.csv", chunksize=200_000):
    partials.append(process_chunk(chunk))
result = pd.concat(partials).groupby(level=0).sum()

Works. Single core. Slow.

Option B: Polars streaming

import polars as pl
(pl.scan_csv("big.csv")
   .with_columns((pl.col("x") ** 2).alias("x2"))
   .group_by("region").agg(pl.col("x2").sum())
   .collect(streaming=True))

Faster, multicore, less code. Often the right tool in 2026.

Option C: Dask

import dask.dataframe as dd
df = dd.read_csv("big.csv", blocksize="64MB")
df["x2"] = df["x"] ** 2
df.groupby("region")["x2"].sum().compute()

Scales beyond one machine if needed.

Option D: Ray Data

import ray.data as rd
def transform(batch):
    batch["x2"] = batch["x"] ** 2
    return batch

ds = rd.read_csv("big.csv").map_batches(transform)
ds.groupby("region").sum("x2").show()

For ML pipelines, Ray Data integrates with PyTorch / TF datasets.

Hands-on lab (2 hours)

1. Write a CPU-bound function; parallelise with ProcessPoolExecutor; measure speedup vs cores.
2. Use shared_memory to share a 100 MB NumPy array across 4 workers (read-only).
3. Install Dask; convert a chunked pandas loop to dd.read_csv(...).compute().
4. Open the Dask dashboard while a job runs; observe task graph.
5. Install Ray; convert one parallel function to @ray.remote.
6. Build a Ray actor for a stateful counter; verify state persists across calls.
7. Bonus: spin up a local Dask Distributed cluster (Client(n_workers=4)); compare to the default scheduler.

Common pitfalls

1. multiprocessing without the __main__ guard → recursive spawn on Windows.
2. Pickling lambdas / closures.
3. Re-pickling huge arrays per task; switch to shared memory or Dask.
4. Using Spark for 10 GB of data — overhead dominates.
5. Forgetting to ray.init() (or Client()).
6. mp.Manager.dict in the hot path — slow proxy calls.
7. Reaching for Dask/Ray when Polars / DuckDB on one machine would do.

Self-check

1. When use Dask vs Ray?
2. Why does shared_memory help with multiprocessing?
3. What's a Ray actor?
4. When is Spark the right tool?
5. Why not Spark for 5 GB of data?

References

• Python multiprocessing docs.
• Dask docs: https://docs.dask.org/.
• Ray docs: https://docs.ray.io/.
• Polars streaming: https://docs.pola.rs/user-guide/concepts/streaming/.
• Scaling Python with Dask, Holden Karau.
• "Bigger Data, Faster" — Wes McKinney blog.