9.1 From Single-Machine Overflow to Partitioned Computing: First Shrink the Working Set, Then Add Machines

Ten sites produced five years of equipment logs, totaling 500 GB in compressed files. pd.read_csv() quickly runs out of memory, but that does not make a cluster the automatic answer. The requested output contains only monthly aggregates by equipment type and incident frequency, so the required working set can be far smaller than the source data.

There's no fixed rule like "16GB use pandas, 100GB use Spark." Tool selection depends on the working set, operators, storage layout, concurrency, deadlines, and the team's operational capabilities.

Learning Objectives

• Distinguish dataset size, working set, and result size;
• Determine which calculations can safely be chunked and merged;
• Use columnar format, column pruning, and filter pushdown to reduce I/O;
• Use evidence to decide whether to continue single-machine, use a parallel framework, or scale up to a cluster.

1. First, identify the bottleneck

A slow query might be limited by:

• Memory: Intermediate tables, hash tables, or sorted data cannot fit in memory;
• I/O: Reading a large number of irrelevant columns and rows;
• CPU: Parses text, compresses data, or executes Python functions and complex expressions;
• Network: remote reading or node-to-node shuffle;
• Scheduling: A large number of small files and tasks;
• Skew: A few keys cause one partition to become a long tail.

Track peak memory, scanned bytes, CPU utilization, I/O throughput, task distribution, and number of output rows. Looking at only the input file size can mislead: a 20GB join might generate several TB of intermediate results, and a 500GB log might end up being read in only a few GB after partition pruning.

2. Working set determines memory pressure

A working set is the data that must be kept simultaneously during a particular execution phase. Reading can be streamed, but this does not mean all operators are streamable.

• Row-level filtering, projection, and row-by-row transformation can typically be executed in parallel across partitions;
• sum, count, min, max can first compute local states and then combine them;
• Precise global sorting, certain window operations, and high-cardinality deduplication may require large-scale reorganization;
• Whether medians, quantiles, and any Python UDF can be chunked depends on the algorithm and implementation.

Block computations must have correct merging rules. For example, the overall mean cannot simply be the average of the block means; instead, merge sum and count:

from io import StringIO
import pandas as pd

csv_data = "mission_type,resources\nscan,10\nscan,20\nscan,90\nrepair,\n,5\n"

def aggregate_chunks(source, chunk_size=2):
    merged = pd.DataFrame(columns=["sum", "count"], dtype=float)
    for chunk in pd.read_csv(source, dtype={"mission_type": "string", "resources": "float64"},
                             usecols=["mission_type", "resources"], chunksize=chunk_size):
        part = chunk.groupby("mission_type", dropna=False)["resources"].agg(["sum", "count"])
        merged = pd.concat([merged, part]).groupby(level=0, dropna=False).sum()
    merged["mean"] = merged["sum"] / merged["count"].where(merged["count"] > 0)
    return merged

merged = aggregate_chunks(StringIO(csv_data))
print(merged.loc["scan", ["sum", "count", "mean"]].tolist())
# [120.0, 3.0, 40.0]
assert pd.isna(merged.loc["repair", "mean"])

This pandas example retains a group for missing keys and excludes missing resource values from the mean denominator. The two chunk means for scan are 15 and 90; averaging them gives the wrong value, 52.5. Combining sum 120 and count three gives 40. Keeping cumulative aggregate state avoids storing every partial table, but memory still grows with group cardinality.

3. First change how data is stored and queried

CSV is great for data exchange but lacks reliable typing, column-level access, and rich statistical information. For data that is analyzed repeatedly, it can be converted to a columnar format with an explicit schema.

Create a shared Parquet fixture in a temporary directory. Install pyarrow, duckdb, and dask[dataframe] in addition to pandas, and run this page’s Python blocks in order:

from pathlib import Path
from tempfile import TemporaryDirectory

work = TemporaryDirectory(prefix="atlas-partitions-")
data_path = Path(work.name) / "missions"
fixture = pd.DataFrame({
    "event_date": pd.to_datetime(["2025-12-31", "2026-01-01", "2026-01-02"]),
    "mission_type": ["scan", "scan", "repair"],
    "resources": [100.0, 10.0, 30.0],
    "year": [2025, 2026, 2026], "month": [12, 1, 1],
})
fixture.to_parquet(data_path, engine="pyarrow", partition_cols=["year", "month"], index=False)
# Keep work alive until the SQL and Dask examples finish.

Columnar formats like Parquet allow the execution engine to read only the needed columns; if the file or row group metadata supports it, filtering can be pushed down to the scan stage and data that cannot match can be skipped. The actual performance gain depends on compression, selectivity, layout, storage, and engine, and cannot guarantee a fixed "5–10x speedup."

SELECT mission_type, avg(resources) AS mean_resources
FROM read_parquet('PATH/year=*/month=*/*.parquet', hive_partitioning = true)
WHERE year = 2026 AND event_date >= TIMESTAMP '2026-01-01'
GROUP BY mission_type;

Querying files directly in a local analysis engine typically uses less memory than loading all the data into a DataFrame.

The SQL uses DuckDB syntax; replace PATH with data_path. The following runs the equivalent query directly:

import duckdb

sql_result = duckdb.sql("""
    SELECT mission_type, avg(resources) AS mean_resources
    FROM read_parquet(?, hive_partitioning = true)
    WHERE year = 2026 AND event_date >= TIMESTAMP '2026-01-01'
    GROUP BY mission_type ORDER BY mission_type
""", params=[str(data_path / "year=*" / "month=*" / "*.parquet")]).fetchall()
print(sql_result)
# [('repair', 30.0), ('scan', 10.0)]

year = 2026 filters directory partitions; event_date filters file contents. A date predicate alone does not guarantee that the engine infers a directory-year predicate. Verify partition pruning and row-group skipping separately through plans and scan statistics.

4. Physical partitions serve common filtering needs

Organizing the directory by fields like date and region allows the engine to skip irrelevant partitions. But having too many partition columns isn't better:

• High-cardinality keys generate massive directory structures and numerous small files;
• When querying without partition predicates, it still requires scanning a large number of files;
• Small files increase list overhead, opening costs, and scheduling overhead;
• Overparallelism during write operations will fragment each partition.

Choose a small, stable set of fields frequently used in coarse-grained filtering. Internal sorting and row-group statistics can further help skip data. Regular compaction requires maintaining atomic publication and concurrent read-write safety.

5. Delayed execution isn't automatically faster

DataFrame frameworks like Dask build operations into a task graph and execute them only when results are needed:

import dask.dataframe as dd

ddf = dd.read_parquet(
    data_path,
    columns=["event_date", "mission_type", "resources"],
    filters=[[("event_date", ">=", pd.Timestamp("2026-01-01"))]],
)
result = ddf.groupby("mission_type")["resources"].agg(["sum", "count"])
final = result.compute()
print(final.sort_index().to_dict("index"))
# {'repair': {'sum': 30.0, 'count': 1}, 'scan': {'sum': 10.0, 'count': 1}}
work.cleanup()

Deferred execution allows the optimizer to perform column pruning, filter pushdown, and task fusion, and avoids premature materialization of intermediate results. But:

• Small partitions can result in a massive task graph;
• An overly large partition can cause multiple parallel tasks to collectively exhaust worker memory;
• Repeatedly calling compute() in a loop destroys optimization opportunities;
• Similarity to the pandas API does not mean identical execution costs or support scope.

First, use the dashboard or run reports to observe tasks, memory, spills, and transfers, rather than guessing a partition number.

This small local run checks results, not throughput. compute() collects the final result into the calling process, so a large output can exhaust client memory even with distributed input. See Dask’s Parquet documentation for metadata and partition-reading behavior.

6. When to Introduce Distributed Processing

Below are decision dimensions, not capacity thresholds:

Question	Better suited for single-machine or embedded engines	Better suited for distributed frameworks
Working set	Can be processed in a streaming fashion or fit within disk spillover	Single-machine time limit or disk bandwidth is unacceptable
Calculation	Scanning, filtering, standard aggregation	Multiple large shuffles, parallel ETL, iterative jobs
Concurrency	One analyst, few tasks	Multiple teams sharing, requiring queues and resource isolation
Data Location	Local or single object-storage region	Data already in cluster/distributed storage
SLA	Acceptable in minutes to hours	Requires higher throughput or stable deadlines
Operations	Hope for fewer components	Already has scheduling, monitoring, security, and on-call capabilities

Scaling up a machine, reducing scans, pre-aggregating data, and optimizing algorithms might be cheaper than adding more nodes. Distributed systems address throughput and capacity through parallelism, but they don't eliminate network and coordination overhead.

7. Take a baseline set of measurements before migration

Choose representative queries, rather than count(*) alone:

Narrow table filtering and aggregation
High-cardinality group by
Joining a large fact table with a small dimension table
Join two large tables
Windows or sorting
The final number of small files written

For each query, report input scale, scanned bytes, peak memory, execution time, cost, and result validation value. Measure cold cache and hot cache separately to avoid mistaking cache benefits for engine performance gains.

Common Misconceptions

• Data exceeds memory requires a cluster: Streaming, disk spilling, and column pruning might be sufficient.
• Switching to Parquet will always provide a fixed multiple improvement: Performance depends on the query and layout.
• Average the chunk means: Different block sizes can lead to incorrect results.
• API similarity enables mechanical migration: Partitioning, indexing, shuffle, and UDF costs differ.

Practice

1. Split an OOM query into four working sets: scan, filter, aggregate, and result.
2. Design combinable states for block variance calculation, rather than averaging block variances.
3. Compare the scan bytes and peak memory usage of CSV full-read versus Parquet column pruning.
4. Compare a single-machine SQL engine with a distributed framework using the same business query, and draw conclusions about the migration.

Summary

“How big is the data” is just the starting point. Decision criteria include the working set at each stage, data movement volume, and service objectives. First, reduce unnecessary reads and intermediate results, then decide whether the complexity of distributed systems is worth bearing.

Next lesson: breaking down the cost structure of distributed batch processing, partitioning enables parallelism, while shuffle, skew, and retries determine whether a job can complete stably.