--- title: "6.6.3 Dask, Spark and Ray" description: "Dask over a tick lake (plan, compute, persist, skewed partitions), Spark and Ray with their real code, and a parameter sweep in parallel." url: "https://pythonparatrading.com/en/python-trading-course/level-6/dask-spark-and-ray/" lang: "en" image: "https://pythonparatrading.com/og/en/python-trading-course/level-6/dask-spark-and-ray.png" --- # 6.6.3 · Dask, Spark and Ray: from one machine to a cluster > A cluster distributes data or distributes computations, and in both cases it pays for moving data between processes. This lesson runs Dask over ten million ticks and measures in bytes the difference between `compute`, `dask.compute` and `persist`, shows why partitioning by symbol unbalances the work, shows the Spark and Ray code (which is not run, and explains why) and distributes a parameter sweep whose best result is pure chance. - Course: [Level 6: Generative AI, LLMs, agents, big data and alternative data](https://pythonparatrading.com/en/python-trading-course/level-6/) - Module: 6.6 · Big data for trading - Duration: 65 min - Prerequisites: [6.6.2 · Parquet, Arrow, Polars streaming and out-of-core DuckDB](https://pythonparatrading.com/en/python-trading-course/level-6/parquet-arrow-polars-streaming-duckdb/), [2.5.4 · Deflated Sharpe Ratio and Probability of Backtest Overfitting (PBO)](https://pythonparatrading.com/en/python-trading-course/level-2/deflated-sharpe-ratio-and-pbo/) ## By the end of this lesson - Explain the common architecture of Dask, Spark and Ray (client, scheduler, workers) and the cost of moving data between them. - Build lazy queries with Dask DataFrame, read their expression tree and choose between `compute`, `dask.compute` and `persist` by measuring what is read. - Tell apart computations that do not cross partitions (`map_partitions`) from those that require redistributing data, and detect unbalanced partitions. - Write the PySpark and Ray equivalent of a Dask computation, and know when each framework pays off. - Parallelize a parameter sweep with `dask.delayed` and correct the best result for multiple testing. ## 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. ```mermaid 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”[^1]. 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. ```python 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") ``` ```output 9,800,152 ticks, 39 files ``` ## Dask DataFrame: pandas by partitions A Dask `DataFrame` is a collection of pandas `DataFrame`s, the **partitions**, ordered along the index and split by rows[^2]. 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: ```python 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) ``` ```output 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: ```python c = ddf.assign(importe=ddf.precio * ddf.tamano) por_simbolo = c.groupby("simbolo") vwap = por_simbolo.importe.sum() / por_simbolo.tamano.sum() vwap.pprint() ``` ```output 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: ```mermaid 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. ```python 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) ``` ```output 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: ```python 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}") ``` ```output 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[^3]. 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: ```python 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"]) ``` ```output 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 ``` *Fig. — Rows per partition of the 9.8-million-tick lake when partitioned by date or by symbol.* Two panels with partitions sorted from largest to smallest. By date, 39 nearly equal partitions between 250,155 and 252,509 rows. By symbol, 200 partitions falling from 238,330 rows (CELE, highlighted) to 2,575, with a median of 39,607. 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` set[^4]. 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: ```python skip title="vwap_spark.py" 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[^4]: ```python skip title="pandas_en_spark.py" 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 `DataFrame`s, 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 them[^5]. On top of it sit Ray Tune (hyperparameter search), Ray Train and Ray Data. The sweep below, written with Ray, would look like this: ```python skip title="barrido_ray.py" 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`[^5]. 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. ```python 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) ``` ```output 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 | Framework | Unit of work | Shines at | Cost | |---|---|---|---| | Polars, DuckDB (lesson 6.6.2) | Query on one machine | Almost everything that fits on one disk | Nothing new | | Dask DataFrame | pandas partition | pandas code that no longer fits; scaling from laptop to cluster | Younger optimizer; `persist` with rough edges | | Dask `delayed` | Python function | Sweeps and tasks with dependencies between them | The scheduler adds per-task overhead | | Spark (PySpark) | Partition in the JVM | Terabyte-scale SQL, existing clusters, Delta and Iceberg | Java, a cluster to maintain, serialization between Python and the JVM | | Ray | Task or actor | ML experiments, hyperparameter search, model serving | Its own services (GCS, raylets), network required | 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. [^1]: Dask, “Dask Best Practices”: , accessed October 23, 2026. [^2]: Dask, “Dask DataFrame” and “Scheduling” (the threaded scheduler is the default for Dask DataFrame): , accessed October 23, 2026. [^3]: Dask, “Dask DataFrames Best Practices”: , accessed October 23, 2026. [^4]: Apache Spark 4.2.0, “Overview” and PySpark “Installation”: and ; project news (). Accessed October 23, 2026. [^5]: Ray 2.59.0, “What's Ray Core?”, “Usage Stats Collection” and the glossary (raylet, GCS): and , accessed October 23, 2026. ## 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. **Solution.** ```python 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())) ``` ```output 200 VWAPs and 7800 volatilities; 33.9 MB read same as before: True ``` A single pass: 33.9 MB, the same as reading for the VWAP alone, because the volatility needs the same columns plus `fecha`, which is not in the files but in the paths. The volatilities match those of the `map_partitions` cell. It is Dask's practical rule: put everything you are going to ask of the same data into one `dask.compute`. 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. **Solution.** ```python 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"]) ``` ```output particiones mediana_MB max_MB one file per partition 39 6.3 6.3 100 MB 3 124.4 124.4 ``` Each file takes up 6.3 MB in memory with pandas (versus less than 1 MB compressed on disk): very small partitions for Dask, where each task has a fixed scheduling cost. `repartition(partition_size="100MB")` merges them into 3 partitions of about 124 MB: Dask estimates the sizes and groups approximately. On a cluster, 3 partitions would leave almost all workers idle; the reasonable size is a compromise between per-task overhead (small partitions) and memory and parallelism (large partitions). The last cell deletes the temporary data. ```python import shutil shutil.rmtree("ticks") print(sorted(p.name for p in Path(".").glob("ticks*"))) ``` ```output [] ``` 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? **Solution.** (a) The daily factors: 60 GB in Parquet, partitioned by date and sorted by symbol within each day, fit on the disk of a single machine, and a nightly computation that only processes the new day reads a few megabytes. Polars lazy with streaming or DuckDB with a memory limit (lesson 6.6.2) are enough, also to recompute the full history now and then. Dask only if the code is already written in pandas and does not fit in memory. (b) The sweeps: 500 independent combinations is a task problem, not a data problem. On one machine, `dask.delayed` or `concurrent.futures` with processes (lesson 0.4.6) use all the cores; if a sweep takes too long, Ray or a Dask cluster in the cloud distribute the same functions across several machines without changing the logic. Spark brings nothing here and adds Java and a cluster to maintain. And with 500 combinations, the best result has to be corrected for multiple testing (Deflated Sharpe, lesson 2.5.4) before using it. ## Self-assessment 1. You need the VWAP per symbol and the number of ticks per symbol from the same lake. With two `compute()` calls 39.1 MB are read; what does `dask.compute(vwap, n)` do? - A. The same as two `compute()` calls: each result has its own graph - B. It optimizes both graphs together and reads each file only once: 33.9 MB - C. It keeps the data in memory forever - D. It computes the results approximately *Answer B: It optimizes both graphs together and reads each file only once: 33.9 MB.* The common tasks (reading each partition) are shared. `persist()` also reads once, but it also leaves the partitions in memory: later queries read 0 bytes at the cost of taking up memory. 2. The daily volatility of each symbol is computed with `map_partitions` without any communication between workers. What makes this possible? - A. Dask reorders the data automatically - B. Volatility is a sum - C. `meta` describes the result - D. Each partition is a file and each file a complete session: all the rows each symbol-day group needs are in the same partition *Answer D: Each partition is a file and each file a complete session: all the rows each symbol-day group needs are in the same partition.* If the computation needed rows from several partitions (for example, returns of a symbol across days), they would have to be redistributed by symbol (shuffle), the most expensive operation in a distributed system. 3. By date, the 39 partitions have between 250,155 and 252,509 rows; by symbol, between 2,575 and 238,330. What is the problem with partitioning by symbol? - A. Partition skew: the job finishes when the largest partition finishes, and the other workers wait - B. None, there are more partitions and therefore more parallelism - C. Dask does not support partitions by string - D. The small partitions do not fit in memory *Answer A: Partition skew: the job finishes when the largest partition finishes, and the other workers wait.* Stock activity is very uneven. The largest partition is six times the median. The ways out: partition by date and sort by symbol within, group small symbols into similar-sized buckets or split the large ones. 4. Why is Ray not run in the course, even though it is installed in the `ml` environment? - A. Because Ray does not work with Python 3.14 - B. Because Ray needs a GPU - C. Because the lesson runs in the main environment and, in the test, `ray.init()` looked up the local IP towards `8.8.8.8`, started its services on the local network address and, with no network, hung waiting for its control service - D. Because Ray is paid software *Answer C: Because the lesson runs in the main environment and, in the test, `ray.init()` looked up the local IP towards `8.8.8.8`, started its services on the local network address and, with no network, hung waiting for its control service.* With a network, the same code finished without problems. The course requires offline, deterministic runs; the sweep runs with `dask.delayed`, which distributes Python functions in the same way, and the Ray code is shown without running it. 5. A sweep of 20 combinations of a moving average crossover over 200 driftless random walks gives a maximum mean Sharpe of 0.028, but the best of the 4,000 strategies has a Sharpe of 0.98 over ten years. Which conclusion is correct? - A. The (40, 250) combination works on that series and should be used - B. The maximum of many tests without signal is high by chance: it has to be corrected for multiple testing (Deflated Sharpe) before believing it - C. Ten years is not enough to estimate a Sharpe - D. Parallelism introduces errors into the computation *Answer B: The maximum of many tests without signal is high by chance: it has to be corrected for multiple testing (Deflated Sharpe) before believing it.* The series have no signal at all by construction. Parallelizing makes it cheaper to try more combinations, and the more you try, the higher the best result by chance (lesson 2.5.4). ## Key points - Dask, Spark and Ray share an architecture (client, scheduler, workers) and a cost: moving data between processes or machines is much slower than reading your own memory. Before distributing, a well-used machine (lesson 6.6.2). - Dask DataFrame is a lazy, optimized collection of pandas `DataFrame`s: two `compute()` calls read 39.1 MB, `dask.compute` of both 33.9 MB and, after `persist()`, 0 MB at the cost of memory. - Computations within a partition (`map_partitions`) communicate nothing; those that cross partitions require a shuffle. Partitioning by date gave almost equal partitions; by symbol, the largest was six times the median. - Spark 4.2 needs Java 17, 21 or 25 and shines with terabyte-scale SQL and existing clusters; Ray distributes tasks and actors for ML experiments; neither is run in the course, because of Java and the network. - Parallelizing a sweep makes it cheaper and multiplies the tests: the best of 4,000 strategies without signal had a Sharpe of 0.98. The Deflated Sharpe is not optional. ## Resources ### Documentation - [Dask DataFrame (official documentation)](https://docs.dask.org/en/stable/dataframe.html) · docs.dask.org - [Dask DataFrames: best practices (partition size, persist)](https://docs.dask.org/en/stable/dataframe-best-practices.html) · docs.dask.org - [Apache Spark 4.2.0: overview and requirements](https://spark.apache.org/docs/latest/) · spark.apache.org - [PySpark: installation (Java 17 or later)](https://spark.apache.org/docs/latest/api/python/getting_started/install.html) · spark.apache.org - [Ray Core: tasks, actors and objects](https://docs.ray.io/en/latest/ray-core/walkthrough.html) · docs.ray.io ### Data - [Level 5 data: tickers of the stock panel (CSV)](https://pythonparatrading.com/curso/datos/nivel-5/panel5_activos.csv) · Download · panel5_activos.csv Previous lesson: [6.6.2 · Parquet, Arrow, Polars streaming and out-of-core DuckDB](https://pythonparatrading.com/en/python-trading-course/level-6/parquet-arrow-polars-streaming-duckdb/) Next lesson: [6.6.4 · Time series databases: TimescaleDB, QuestDB, ClickHouse, ArcticDB and kdb+](https://pythonparatrading.com/en/python-trading-course/level-6/time-series-databases-and-arcticdb/)