When Distributed Processing Is Needed

A team has a nightly job that aggregates about forty million order rows. It takes thirty minutes on one machine, and the plan for next quarter is to move it to a cluster. The reasoning is familiar and rarely examined: the data is growing, clusters are what you use for data at scale, and the platform team already runs one. The example is invented, and the decision may well be right โ€” but it is being made on a premise that stopped being reliable some years ago, which is that a single machine is small.

This article is about that judgment: what distribution actually buys, what it costs, which conditions genuinely force it, and how much a single node handles today. Hardware ceilings and engine capabilities change; confirm the values relevant to the machine and version you can use.

The problem the phrase “big data” pointed at has moved

When the vocabulary of big data formed, the constraint was concrete: datasets exceeded what one affordable machine could hold or read in reasonable time, so the work had to be spread across many cheap machines. Two things have changed since.

  • One machine got much larger. AWS documents its U7i high-memory instances as offering up to 32 TiB of memory and up to 1,920 vCPUs, in models such as u7i-12tb.224xlarge (896 vCPUs, 12 TiB). These exist for in-memory databases rather than for analytics, but they set the ceiling: a dataset of a few terabytes is no longer automatically a multi-machine problem.
  • Single-node engines stopped requiring the data to fit in memory. DuckDB’s documentation states that it supports “larger-than-memory workloads, i.e., it is able to process datasets that are larger than the available system memory (also known as out-of-core processing),” spilling to disk for blocking operators such as GROUP BY, JOIN, ORDER BY, and windowed aggregation.

Add columnar storage, vectorized execution, and the fact that most analytical queries read a few columns of a few partitions rather than everything, and the practical ceiling of one machine is far above where most people place it. What often grows faster than data volume is the number of concurrent users, the number of pipelines, and the number of sources โ€” which are different problems with different solutions.

The uncomfortable benchmark

In 2015 Frank McSherry, Michael Isard, and Derek Murray proposed a metric they called COST โ€” “the Configuration that Outperforms a Single Thread.” Their definition: “The COST of a given platform for a given problem is the hardware configuration required before the platform outperforms a competent single-threaded implementation.” The point of measuring it is stated in their introduction as a question: “To what degree are these systems truly improving performance, as opposed to parallelizing overheads that they themselves introduce?”

Surveying data-parallel systems published at major systems conferences, they found many with “a surprisingly large COST, often hundreds of cores, or simply underperform one thread for all of their reported configurations” โ€” several had unbounded COST, meaning that among the configurations reported for them, none beat a competent single-threaded implementation on a high-end laptop. Their epigraph, quoting Paul Barham, is the sentence worth keeping: “You can have a second computer once you’ve shown you know how to use the first one.”

The paper is about graph processing systems in 2015, not about your warehouse in 2026, and citing it as proof that clusters are useless would misread it. What transfers is the discipline: a scaling curve shows how a system behaves as you add resources, not whether the absolute performance is good. Twice as fast on sixteen machines is not impressive if one machine with a better data layout is faster than all of them.

What actually forces distribution

Four conditions are the usual reasons to reach for more than one machine. Capacity requires a measured single-node limit; concurrency can call for workload isolation, availability for replication, and elasticity for a different operating arrangement. Whether a condition requires multiple machines depends on the service target and the alternatives below. Volume alone is rarely the first one to bite.

ConditionWhat it looks likeWhy one node cannot do it
Throughput or capacity beyond one available machineThe job must read tens of terabytes within a fixed window, or its CPU-bound transformation, its in-memory operator state, or its temporary disk exceeds what one node you can actually use providesAggregate bandwidth, cores, memory, and scratch space scale with machines; a single node’s are bounded โ€” by what your budget, region, and operations allow, not by the largest instance in a catalog. This forces distribution only when the work can be split and the distributed version meets the same accuracy, deadline, and cost, which is a measurement rather than an assumption
ConcurrencyHundreds of simultaneous queries or many pipelines competingA single node runs queries in parallel and can be given resource limits, so this forces distribution only once one node’s throughput or isolation targets are actually missed. Separate compute per workload is the usual answer, and that is workload isolation rather than distributing one query
AvailabilityThe work must survive a machine failure without waiting for a restartOne node is one failure domain
Elasticity of costA ten-minute daily peak against a mostly idle dayAlso a choice rather than a requirement: a single machine can be started for the run and stopped after it. Managed clusters make that automatic and finer-grained, which is why it is usually bought rather than built

Notice that three of the four are not about data size. That is the practical shift: when distribution is bought for isolation, availability, or elasticity, those are good reasons. It is worth keeping three different things apart while deciding, because they need different architectures: splitting one query across machines, giving each workload its own compute, and replicating for failover. “The dataset is large” is the reason people give, and it is the one that most often does not hold up when measured. For the mechanics of how distributed engines divide the work, see massively parallel processing.

What parallelism costs

Splitting work across machines adds expense in two layers. Some costs come with parallelism itself and exist on one multi-core node too: scheduling tasks, choosing their size, and waiting for the slowest one. Others come only from crossing hosts: moving data over a network, remote failures, and coordinating state between machines. The list below marks which is which.

  • Moving data between machines (hosts only). An operation whose required grouping the current placement does not satisfy โ€” a join on a key the data is not already partitioned by, a global aggregation, a sort โ€” needs an exchange, and across machines that is network traffic. How much moves depends on placement and plan: a join whose inputs already share the required partitioning can skip the exchange, and a local partial aggregation shrinks what a global one has to send. Spark’s tuning guide notes that shuffle operations build a hash table within each task that “can often be large,” and that data locality “can have a major impact” on performance. Read the bytes exchanged, serialized, and spilled from the actual plan and metrics rather than from the operation’s name; a single node pays for regrouping in memory and disk, not in network.
  • Serialization (mostly hosts, sometimes local). Data that crosses a machine boundary must be encoded and decoded; on one node, spilling to disk or handing data between processes can require it too, while shared memory can avoid it. The same guide calls serialization something that “plays an important role in the performance of any distributed application” and often “the first thing you should tune.”
  • Coordination (both, and more across hosts). Scheduling tasks, tracking their status, and handling stragglers costs time relative to a loop on one thread โ€” and a multi-core engine on one node has the same problem in smaller form, which is why DuckDB parallelizes over row groups and lets the thread count be tuned. Across machines the tracking is remote and a task can be lost with its host. Spark’s recommendation of “2-3 tasks per CPU core” is that tuning problem at cluster scale.
  • Skew. Parallel work finishes when the slowest partition finishes. One popular key, one enormous customer, one null-heavy join column, and forty machines wait for one โ€” see partitioning, shuffles, and data skew.
  • Operations and expertise. A cluster brings cluster sizing, version upgrades, dependency management, failure modes that only appear under load, and a debugging experience where the stack trace is on another machine. That burden is real whether or not it appears on an invoice.

None of this argues against clusters. It argues that the comparison is not “one machine, slowly” against “many machines, quickly” but against “many machines, quickly, minus the overheads the distribution itself introduced, plus the work of running it.”

How far one node goes, in practice

A single-node engine reading columnar files handles more than most estimates assume, because the data that matters is smaller than the data that exists:

  • Partition pruning removes whole files before reading, so “three years of history” is often “eleven days” for the query at hand.
  • Predicate pushdown and column projection mean a 200-column table read for four columns costs roughly the columns actually needed โ€” the four in the output plus any used by filters and joins โ€” rather than all 200; the ratio of columns is not the ratio of bytes or time.
  • Out-of-core processing removes the requirement that the working set fit in RAM, within documented limits โ€” DuckDB notes that several blocking operators in one query can still exhaust memory, and that some aggregate functions cannot spill at all.

The honest way to find the boundary is to measure it rather than to reason about it: run the real query on the real data on one large machine, and see. Single-node analytics with DuckDB, Polars, and Arrow covers the tooling, and working with tabular data at scale covers the techniques that keep it viable.

A decision order that avoids the common mistake

  1. State the requirement in time, not in size. “This must finish within twenty minutes of the source being available” is a testable constraint; “we have big data” is not.
  2. Measure the honest baseline. One machine, columnar files, a sensible partition layout, and the real query. A “we need a cluster” conversation that has never seen this number is arguing from a guess.
  3. Fix the data layout before adding machines. Small files, no partitioning, row formats, and reading columns nobody uses are the usual reasons a job is slow โ€” and distribution multiplies those costs rather than removing them. File compaction is often the cheapest speedup available.
  4. Name which of the four conditions applies. Throughput or capacity, concurrency, availability, or elasticity โ€” with the number that shows it.
  5. Compare the whole cost, including the operating burden and the skills required, not just the compute bill. That is ordinary total cost of ownership reasoning applied to an architecture choice.

Then leave the decision alone until something moves it. Revisit when measured headroom shrinks โ€” processing time against the deadline, peak memory, temporary disk, concurrency, or recovery targets trending toward their limits โ€” or when the workload’s shape or the service requirements change. A job at 18 minutes against a 20-minute deadline is moved by 12 percent growth, not by an order of magnitude, if processing time grows in proportion to the workload: 18 ร— 1.12 = 20.16 minutes. Periodic checks should validate those assumptions; they need not trigger a migration, and a review with no changed measurement should change nothing.

And the reverse case deserves the same discipline. A team already running a cluster that works, with staff who know it, should not move a stable workload to a single node because a blog post said so. The cost of a migration is real, and “it would run on one machine” is not by itself a reason to make it.

References

Hardware ceilings and engine capabilities change, so confirm current values before deciding.


Discover more from Insightful Data Lab

Subscribe to get the latest posts sent to your email.

Similar Posts

Questions, corrections, or additional insights?

This site uses Akismet to reduce spam. Learn how your comment data is processed.