Dask, Spark and Ray: from one machine to a cluster.

When you need more than one machine

Lesson 6.6.2 took a single machine a long way: 21 million ticks with Polars in streaming mode and DuckDB with 300 MB of memory. There are three situations in which that stops being enough:

• The data does not even fit on one machine's disk, or reading it with a single disk and a single network card takes too long: years of L3 books for a global universe, full option chains (lesson 6.6.1).
• The computation is huge and splits well: thousands of backtests, a parameter sweep, a cross-validation with many folds, each one independent of the others.
• Several people or processes share the same resources: a team that launches jobs over the same lake.

For the first, distributed data frameworks (Dask DataFrame, Spark) split a table into chunks and distribute the chunks across machines. For the second, distributed task frameworks (Dask delayed, Ray) distribute Python functions. Both share the same architecture: one process that schedules and several that do the work.

flowchart TB
  C[Client: your code] --> S[Scheduler]
  S --> W1[Worker 1]
  S --> W2[Worker 2]
  S --> W3[Worker 3]
  W1 --> L[(Parquet lake)]
  W2 --> L
  W3 --> L

In Dask the scheduler is just the scheduler; in Spark, it is the driver, and the workers, executors; in Ray, each node has a raylet that schedules locally and a central service (GCS) keeps the cluster state. The names change; the idea does not. And it has a cost worth keeping in mind from the start: moving data between processes or machines requires serializing it and sending it over the network, which is much slower than reading it from your own memory. The Dask documentation says it bluntly: parallelism adds complexity and overhead, it is sometimes necessary for large problems, “but often it is not”11Note 1 · Dask, “Dask Best Practices”: https://docs.dask.org/en/stable/best-practices.html, accessed October 23, 2026..

This lesson runs Dask on one machine (its local threaded scheduler), which is the same as it would do on a cluster with another scheduler, and shows the Spark and Ray code without running it, for the reasons explained in each section. The data is synthetic ticks generated with the same method as in lesson 6.6.2, smaller: the 39 NYSE sessions of January and February 2025 (January 9 was an extraordinary closure, a national day of mourning for former President Carter, which the exchange_calendars XNYS calendar records as a holiday) for 200 stocks, one file per session.

from pathlib import Path

import exchange_calendars as xcals
import numpy as np
import pandas as pd
import pyarrow as pa
import pyarrow.parquet as pq

tickers = np.array(
    sorted(pd.read_csv("panel5_activos.csv")["ticker"])[:200])
cal = xcals.get_calendar("XNYS")
sesiones = cal.sessions_in_range("2025-01-02", "2025-02-28")
rng = np.random.default_rng(663)
actividad = np.exp(rng.normal(np.log(1000), 0.8, 200))
ultimo = np.exp(rng.uniform(np.log(10), np.log(400), 200))

def ticks_de_una_sesion(sesion):
    global ultimo
    abre = cal.session_open(sesion).value // 1000   # µs, UTC
    cierra = cal.session_close(sesion).value // 1000
    n = rng.poisson(actividad)
    sim = np.repeat(np.arange(200), n)
    N = len(sim)
    inicio = np.repeat(np.cumsum(n) - n, n)
    ts = rng.integers(abre, cierra - 100_000, N)
    ts = ts[np.lexsort((ts, sim))] + (np.arange(N) - inicio)
    paso = rng.normal(0, 1, N) * np.repeat(
        0.02 / np.sqrt(np.maximum(n, 1)), n)
    c = np.concatenate([[0.0], np.cumsum(paso)])
    camino = c[1:] - c[np.cumsum(n) - n].repeat(n)
    precio = np.round(np.repeat(ultimo, n) * np.exp(camino), 2)
    ultimo = np.where(n > 0, precio[np.cumsum(n) - 1], ultimo)
    return pa.table({
        "ts": pa.array(ts, pa.timestamp("us", tz="UTC")),
        "simbolo": pa.array(tickers).take(sim),
        "precio": precio,
        "tamano": rng.geometric(0.02, N).astype(np.int32)})

filas = 0
for s in sesiones:
    t = ticks_de_una_sesion(s)
    ruta = Path(f"ticks/fecha={s.date()}/parte-0.parquet")
    ruta.parent.mkdir(parents=True)
    pq.write_table(t, ruta, compression="zstd")
    filas += t.num_rows
print(f"{filas:,} ticks, {len(sesiones)} files")

9,800,152 ticks, 39 files

Dask DataFrame: pandas by partitions

A Dask DataFrame is a collection of pandas DataFrames, the partitions, ordered along the index and split by rows22Note 2 · Dask, “Dask DataFrame” and “Scheduling” (the threaded scheduler is the default for Dask DataFrame): https://docs.dask.org/en/stable/dataframe.html, accessed October 23, 2026.. Each operation on it translates into a pandas operation on each partition plus, when needed, a step that combines the results. With read_parquet over a partitioned lake, each file is one partition:

import dask
import dask.dataframe as dd

ddf = dd.read_parquet("ticks", columns=["simbolo", "precio",
                                        "tamano", "fecha"])
print(type(ddf).__name__, "with", ddf.npartitions, "partitions")
print(ddf.dtypes)
ddf.partitions[0].compute().head(3)

DataFrame with 39 partitions
simbolo      string
precio      float64
tamano        int32
fecha      category
dtype: object
  simbolo  precio  tamano       fecha
0    BAGO   34.29     164  2025-01-02
1    BAGO   34.27      20  2025-01-02
2    BAGO   34.25      46  2025-01-02

Creating ddf has not read any data: only the metadata to know the schema and the number of partitions. The fecha column comes from the folder names as category, just as with pandas (lesson 1.4.7). ddf.partitions[0] selects the first partition and .compute() turns it into a pandas DataFrame: the January 2 session.

A lazy query

Like Polars lazy, Dask builds a plan before running anything. The VWAP of each symbol over the two months:

c = ddf.assign(importe=ddf.precio * ddf.tamano)
por_simbolo = c.groupby("simbolo")
vwap = por_simbolo.importe.sum() / por_simbolo.tamano.sum()
vwap.pprint()

OpAlignPartitions: op='__truediv__'
  Sum: observed=True chunk_kwargs={'numeric_only': False} aggregate_kwargs={'numeric_only': False} _slice='importe''simbolo'
    Projection: columns=['simbolo', 'importe']
      Assign: importe=
        ReadParquetFSSpec: path='ticks' columns=['simbolo', 'precio', 'tamano', 'fecha']
        Mul:
          Projection: columns='precio'
            ReadParquetFSSpec: path='ticks' columns=['simbolo', 'precio', 'tamano', 'fecha']
          Projection: columns='tamano'
            ReadParquetFSSpec: path='ticks' columns=['simbolo', 'precio', 'tamano', 'fecha']
  Sum: observed=True chunk_kwargs={'numeric_only': False} aggregate_kwargs={'numeric_only': False} _slice='tamano''simbolo'
    Projection: columns=['simbolo', 'tamano']
      Assign: importe=
        ReadParquetFSSpec: path='ticks' columns=['simbolo', 'precio', 'tamano', 'fecha']
        Mul:
          Projection: columns='precio'
            ReadParquetFSSpec: path='ticks' columns=['simbolo', 'precio', 'tamano', 'fecha']
          Projection: columns='tamano'
            ReadParquetFSSpec: path='ticks' columns=['simbolo', 'precio', 'tamano', 'fecha']

It is the tree as written, from top to bottom: a division (__truediv__) of two group sums, each over a projection of the table with the importe (notional) column added, which in turn comes from multiplying precio by tamano read from Parquet. Before running it, Dask optimizes it as Polars does: it recognizes that both branches read the same thing and merges them, pushes the projection down to the reader (fecha is not needed) and fuses the operations done partition by partition into a single task. Then it turns it into a task graph: one task per partition and operation, plus the combination tasks.

The group sum is the most common pattern of a distributed computation: each partition computes its partial sum per symbol (a small table of 200 rows) and then they are combined in a tree, eight at a time by default (split_every=8), down to a single result. Only the partial sums travel, never the ticks:

flowchart TB
  P1[Partition 1] --> S1[Partial sum]
  P2[Partition 2] --> S2[Partial sum]
  P3[Partition 39] --> S3[Partial sum]
  S1 --> C[Combine]
  S2 --> C
  S3 --> C
  C --> R[VWAP per symbol]

compute, dask.compute and persist

compute() runs the graph and returns a pandas object. If you then ask for another result over the same data, everything runs again, reading included. There are two ways to avoid it: ask for both results at once with dask.compute(a, b), which shares the common tasks, or materialize the data in the workers' memory with persist(), which returns another Dask DataFrame whose partitions are already computed. To see the difference without a stopwatch, the cell again uses a file system that counts what is read (as in lesson 6.6.2) and computes two things: the VWAP per symbol and the number of ticks per symbol.

import os

from fsspec.implementations.local import LocalFileSystem

class Contador(LocalFileSystem):
    """Local disk that records bytes read and files opened."""
    bytes_leidos = 0
    aperturas = 0

    def _open(self, path, mode="rb", **kw):
        f = super()._open(path, mode, **kw)
        self.aperturas += 1
        leer = f.read
        def read(n=-1):
            datos = leer(n)
            self.bytes_leidos += len(datos)
            return datos
        f.read = read
        return f

disco = Contador()
lago = dd.read_parquet(os.path.abspath("ticks"), filesystem=disco,
                       columns=["simbolo", "precio", "tamano"])
c = lago.assign(importe=lago.precio * lago.tamano)

def consultas(df):
    g = df.groupby("simbolo")
    return g.importe.sum() / g.tamano.sum(), df.simbolo.value_counts()

def medir(nombre, accion):
    disco.bytes_leidos = disco.aperturas = 0
    resultado = accion()
    filas.append({"estrategia": nombre, "aperturas": disco.aperturas,
                  "MB_leidos": round(disco.bytes_leidos / 1e6, 1)})
    return resultado

filas = []
vwap, n = consultas(c)
medir("compute separately", lambda: (vwap.compute(), n.compute()))
r1 = medir("dask.compute together", lambda: dask.compute(vwap, n))
p = medir("persist", lambda: c.persist())
r2 = medir("after persist",
           lambda: dask.compute(*consultas(p)))
print("same results:",
      np.allclose(r1[0], r2[0].reindex(r1[0].index), rtol=1e-12),
      r1[1].sort_index().equals(r2[1].sort_index()))
pd.DataFrame(filas)

same results: True True
              estrategia  aperturas  MB_leidos
0     compute separately        156       39.1
1  dask.compute together         78       33.9
2                persist         78       33.9
3          after persist          0        0.0

The figure in the header sums it up. With separate compute() calls, each query reads the lake on its own: 156 file opens (aperturas; two per file and pass, four passes in total counting the two branches the optimizer does not merge across different queries) and 39.1 MB. It is not double one pass because the optimizer pushes the projection down: value_counts only reads the simbolo column. With dask.compute(vwap, n), Dask optimizes both graphs together and reads each file in a single pass: 33.9 MB. persist() reads the same once and keeps the partitions in memory; the two following queries do not touch the disk (0.0 MB) and return the same (True True; the VWAP is compared with np.allclose because the partial sums are combined in a different order and may differ in the last decimal place).

When to use each one:

• dask.compute(a, b, …) when you know in advance all the results you want: a single pass, without taking up memory with the data.
• persist() when you are going to explore interactively over a filtered subset that fits in the cluster's memory: you pay for the read once. Persisted data takes up memory until the variable is deleted.
• compute() at the end, to bring a small result to the local machine. Calling compute() on a large DataFrame tries to fit all of it into the client's memory, which is exactly what you wanted to avoid.

Computations that do not cross partitions

If a computation only needs the rows of one partition, Dask distributes it without any communication between workers. map_partitions applies a pandas function to each partition as is. Here, the realized volatility of each symbol in each session, which only depends on that session's ticks:

def vol_diaria(df):
    r = np.log(df["precio"]).groupby(
        [df["simbolo"], df["fecha"]], observed=True).diff()
    return (r.pow(2).groupby([df["simbolo"], df["fecha"]],
                             observed=True).sum() ** 0.5
            ).rename("vol").reset_index()

meta = pd.DataFrame({"simbolo": pd.Series(dtype="str"),
                     "fecha": pd.Series(dtype=ddf.fecha.dtype),
                     "vol": pd.Series(dtype="float64")})
vol = ddf.map_partitions(vol_diaria, meta=meta).compute()
print(len(vol), "symbol-day pairs;",
      f"median daily volatility: {vol['vol'].median():.4f}")

7800 symbol-day pairs; median daily volatility: 0.0203

7,800 symbol-day pairs (200 × 39) and a median daily volatility of 2.03%: the volatility the prices were generated with (2%), recovered from the ticks. meta tells Dask which columns and types the function will return, so that it can keep building the plan without running it.

The computation is correct only because each session is entirely inside one partition: the partitions match the files, and the files, the days. If you wanted the volatility of each symbol over the whole period with the return between the last tick of one day and the first of the next, a symbol's rows would be spread across 39 partitions and would have to be reorganized by symbol: a redistribution (shuffle), in which each worker sends every other worker the rows that belong to it. It is the most expensive operation of any distributed system, and the reason why the way you partition matters more than the number of machines.

Balanced partitions

The size of the partitions also matters. The Dask guide recommends partitions that fit comfortably in a worker's memory, about 100 MB each, and warns that with 1 GB partitions and ten cores the process can use at least 10 GB³. Just as important is that they are similar to each other: the computation finishes when the slowest partition finishes. The cell compares the two natural ways of partitioning this lake, by date or by symbol, counting the rows of each partition:

por_fecha = ddf.fecha.value_counts().compute()
por_simbolo = r1[1]

def resumen(s):
    return {"particiones": len(s), "mínimo": s.min(),
            "mediana": int(s.median()), "máximo": s.max(),
            "máx/mediana": round(s.max() / s.median(), 1)}

pd.DataFrame([resumen(por_fecha), resumen(por_simbolo)],
             index=["by date", "by symbol"])

           particiones  mínimo  mediana  máximo  máx/mediana
by date             39  250155   251314  252509          1.0
by symbol          200    2575    39607  238330          6.0

By date, the 39 partitions have between 250,155 and 252,509 rows: almost identical, because every day ticks are generated for all symbols. By symbol, from 2,575 to 238,330: the largest is six times the median and ninety times the smallest, because stock activity is very uneven (in the data, a lognormal; in the market, the same or worse). With 200 partitions by symbol spread across eight workers, the one that gets CELE and two other large ones will work much more than the rest while the others wait. This is partition skew, and in market data it appears as soon as you partition by asset. The usual ways out: partition by date and sort by symbol within (as in lesson 6.6.2), group the small symbols into buckets of similar size or split the large ones.

Spark: the cluster standard

Apache Spark is the most widespread distributed data framework: an engine written in Scala that runs on the Java virtual machine, with APIs for Python (PySpark), SQL, Scala, Java and R. Its model is the same as Dask's: a lazy DataFrame split into partitions, a query optimizer and a scheduler that distributes tasks among executors. The latest released version is 4.2.0, from July 14, 2026; according to its documentation, it runs on Java 17, 21 or 25 and Python 3.10 or later, and PySpark needs Java 17 or later with JAVA_HOME set44Note 4 · Apache Spark 4.2.0, “Overview” and PySpark “Installation”: https://spark.apache.org/docs/latest/ and https://spark.apache.org/docs/latest/api/python/getting_started/install.html; project news (https://spark.apache.org/news/index.html). Accessed October 23, 2026.. The course environment has no Java (and pyspark did not resolve during installation), so the code is not run. It is the equivalent of the VWAP per symbol and of the tick-by-tick volatility with a window:

from pyspark.sql import SparkSession, Window
from pyspark.sql import functions as F

spark = (SparkSession.builder.appName("ticks")
         .master("local[4]")            # or the cluster URL
         .getOrCreate())

ticks = spark.read.parquet("ticks")     # discovers fecha=… as a column
vwap = (ticks.groupBy("simbolo")
        .agg((F.sum(F.col("precio") * F.col("tamano"))
              / F.sum("tamano")).alias("vwap"),
             F.count("*").alias("ticks")))

w = Window.partitionBy("simbolo").orderBy("ts")
ret = ticks.withColumn(
    "ret", F.log(F.col("precio") / F.lag("precio").over(w)))
resumen = ret.agg(F.count("ret"), F.sum(F.col("ret") ** 2))

vwap.write.mode("overwrite").parquet("vwap_spark")  # runs here
print(resumen.collect())
spark.stop()

As in Dask and Polars, nothing is computed until an action (write, collect, count); until then Spark only builds the plan, which can be seen with vwap.explain(). Window.partitionBy("simbolo") forces the ticks to be redistributed by symbol among the executors (the shuffle from the previous section). Spark also offers a pandas-shaped API, pyspark.pandas, which follows the specification of the latest pandas version⁴:

import pyspark.pandas as ps

t = ps.read_parquet("ticks")
t["importe"] = t["precio"] * t["tamano"]
g = t.groupby("simbolo")
vwap = g["importe"].sum() / g["tamano"].sum()
print(vwap.sort_index().head(3))

When Spark and not Dask: when there is already a Spark cluster (Databricks, EMR, Dataproc or your own) and a team that maintains it, when the work is mostly terabyte-scale SQL or when you need its ecosystem (reading and writing Delta Lake and Iceberg, lesson 6.6.5). Dask is Python from start to finish: the partitions are pandas DataFrames, your own functions run as is and the same code runs on a laptop and on a cluster; in exchange, its optimizer is younger and its cluster ecosystem, smaller.

Ray and dask.delayed: parallelizing experiments

Many of a researcher's problems are not about big data, but about many independent computations: trying each parameter combination, each fold of a purged cross-validation (lesson 5.4.2) or each seed of a model. That does not need a distributed DataFrame; distributing function calls is enough.

Ray is a framework designed for that. Its core has three pieces: a task is a function decorated with @ray.remote that is launched with .remote() and returns a reference to a future result; an actor is a stateful class that lives in a process; and a shared object store keeps the results so that other tasks can use them without copying them55Note 5 · Ray 2.59.0, “What's Ray Core?”, “Usage Stats Collection” and the glossary (raylet, GCS): https://docs.ray.io/en/latest/ray-core/walkthrough.html and https://docs.ray.io/en/latest/cluster/usage-stats.html, accessed October 23, 2026.. On top of it sit Ray Tune (hyperparameter search), Ray Train and Ray Data. The sweep below, written with Ray, would look like this:

import ray

ray.init(num_cpus=4)          # local; ray.init("auto") on a cluster

@ray.remote
def sharpe_cruce_remoto(rapida, lenta):
    return sharpe_cruce(rapida, lenta)

refs = [sharpe_cruce_remoto.remote(a, b)
        for a in (5, 10, 20, 40) for b in (50, 100, 150, 200, 250)]
res = pd.DataFrame(ray.get(refs))   # waits for the 20 tasks
ray.shutdown()

Ray is installed in the course's ml environment (it comes with NeuralForecast, level 5), but it is not run in this lesson for two reasons. The first, that this lesson runs in the main environment, where Dask is and Ray is not. The second, that its startup is not compatible with the course's offline execution: in the test done for this lesson (Ray 2.59.0), ray.init() tried to find out the local IP by opening a connection to 8.8.8.8:53, started its services on the local network address rather than on 127.0.0.1, and with no network the startup hung waiting for its own control service (GCS). It also writes its session files to /tmp/ray, outside the lesson folder. With a network, the same code finished without problems. Ray collects anonymous usage statistics in some cases; its documentation explains how to disable it with RAY_USAGE_STATS_ENABLED=0⁵.

Dask has its own way of distributing functions, dask.delayed, which is run here. The experiment: a moving average crossover (long if the fast one is above the slow one, short otherwise) over 200 driftless 10-year random walks, with 20 parameter combinations. The known answer is that no combination has an edge: in a random walk, the past says nothing about the future.

from dask import delayed

def sharpe_cruce(rapida, lenta, n_series=200, dias=2520):
    """Moving average crossover on 200 driftless random walks."""
    rng = np.random.default_rng(2026)   # always the same series
    r = rng.normal(0.0, 0.01, (dias, n_series))
    p = pd.DataFrame(np.exp(np.cumsum(r, axis=0)))
    senal = np.where(p.rolling(rapida).mean()
                     > p.rolling(lenta).mean(), 1.0, -1.0)
    pos = pd.DataFrame(senal).shift(1)      # trade on the next day
    rend = (pos * np.expm1(r)).iloc[lenta:]
    sr = rend.mean() / rend.std() * np.sqrt(252)
    return {"rapida": rapida, "lenta": lenta,
            "sharpe_medio": sr.mean(), "sharpe_max": sr.max()}

tareas = [delayed(sharpe_cruce)(a, b)
          for a in (5, 10, 20, 40) for b in (50, 100, 150, 200, 250)]
res = pd.DataFrame(dask.compute(*tareas, scheduler="threads"))
print(len(tareas), "tasks")
mejor = res.loc[res["sharpe_max"].idxmax()]
print(f"best series of the {20 * 200:,}: Sharpe"
      f" {mejor['sharpe_max']:.2f} with ({mejor['rapida']:.0f},"
      f" {mejor['lenta']:.0f})")
res.sort_values("sharpe_medio", ascending=False).round(3).head(4)

20 tasks
best series of the 4,000: Sharpe 0.98 with (40, 250)
    rapida  lenta  sharpe_medio  sharpe_max
3        5    200         0.028       0.808
13      20    200         0.024       0.930
10      20     50         0.024       0.736
4        5    250         0.019       0.917

The 20 tasks run in parallel with the threaded scheduler (NumPy and pandas release the GIL in a good part of their operations; for pure Python functions scheduler="processes" is better). The results do not depend on the scheduler: each task generates the same series with its own seed.

What the numbers say is a level 2 lesson in another context. The best mean Sharpe of the 20 combinations is 0.028, compatible with zero, as it should be. But among the 4,000 strategies tested (200 series × 20 combinations), the best one (fast of 40 sessions, slow of 250) has a Sharpe of 0.98 over ten years, with no signal at all behind it. Parallelizing a sweep makes it cheap and that is why it tempts you to test more: the more tests, the higher the maximum by chance. The Deflated Sharpe (lesson 2.5.4) is the mandatory correction before believing the best result of a sweep, whether it runs on a laptop or on a cluster.

What to choose

The decision order that follows from the lesson: first, a well-used machine (format, plan, streaming); then, Dask if the code is already pandas and more memory or more cores are needed; Spark if the organization already has it; Ray or delayed to distribute experiments. In all three cases, the way the data is partitioned (by date, without skew, with partitions of a reasonable size) decides more than the number of machines.

Exercises

1. Compute, with a single pass over the lake, the VWAP per symbol and the daily volatility from the map_partitions cell, and check with the counter how many bytes are read.

lago_f = dd.read_parquet(
    os.path.abspath("ticks"), filesystem=disco,
    columns=["simbolo", "precio", "tamano", "fecha"])
c_f = lago_f.assign(importe=lago_f.precio * lago_f.tamano)
disco.bytes_leidos = 0
v, vol2 = dask.compute(consultas(c_f)[0],
                       c_f.map_partitions(vol_diaria, meta=meta))
print(len(v), "VWAPs and", len(vol2), "volatilities;",
      f"{disco.bytes_leidos / 1e6:.1f} MB read")
print("same as before:", np.allclose(
    vol2.set_index(["simbolo", "fecha"])["vol"].sort_index(),
    vol.set_index(["simbolo", "fecha"])["vol"].sort_index()))

200 VWAPs and 7800 volatilities; 33.9 MB read
same as before: True

2. How much memory does each partition of ddf take up? Repartition the lake into partitions of about 100 MB with repartition(partition_size="100MB") and compare.

def mb_por_particion(df):
    m = df.memory_usage_per_partition(deep=True).compute() / 1e6
    return {"particiones": len(m), "mediana_MB": round(m.median(), 1),
            "max_MB": round(m.max(), 1)}

grandes = ddf.repartition(partition_size="100MB")
pd.DataFrame([mb_por_particion(ddf), mb_por_particion(grandes)],
             index=["one file per partition", "100 MB"])

                        particiones  mediana_MB  max_MB
one file per partition           39         6.3     6.3
100 MB                            3       124.4   124.4

import shutil

shutil.rmtree("ticks")
print(sorted(p.name for p in Path(".").glob("ticks*")))

[]

3. A team of three people has ten years of 1-minute bars for 3,000 stocks (about 3 billion rows, on the order of 60 GB in Parquet) and wants to (a) compute daily factors every night and (b) launch sweeps of 500 parameter combinations of a model. They have no cluster. Which tools do you propose for each?