0%
Big Data Engines: Spark, Lakehouse, and Streaming at Scale
The Shape of the Problem
Distributed Execution
Spark
Streaming at Scale
Pipelines in Production
Trino and Federated Query
A distributed SQL engine with no storage of its own is a different animal from a warehouse. Nothing is cached for you, nothing is pre sorted for you, and every query starts by asking other systems for bytes. Understanding where the work lands is the whole skill.
The three roles
There is one coordinator, many workers, and a set of connectors.
The coordinator accepts the query, parses it, plans it, splits the plan into stages, assigns tasks to workers, and tracks their progress. It does not read data and it does not process rows. It is a scheduler with a SQL parser attached.
Workers execute tasks. They read from connectors, run operators, exchange data with each other, and return results to the coordinator. All the real work happens here, which is why the worker count is the dial that matters for throughput.
Connectors translate between the engine and a data source. A connector knows how to list what tables exist, how to produce splits that can be read in parallel, and how much of a query it can take on itself. That last capability is where most of the performance in a federated query lives.
From SQL to a distributed plan
The path is fixed and worth knowing by name, because error messages refer to its stages.
Parsing turns text into a syntax tree. Analysis resolves names against catalogs, which is the first point at which a remote system is contacted, usually for metadata only. Logical planning produces a tree of relational operators. Optimisation rewrites that tree, which includes pushing filters down, reordering joins, and deciding what the connector will be asked to do. Fragmentation cuts the optimised plan at every exchange boundary into stages. Scheduling assigns each stage's tasks to workers.
The interesting thing about this list is how much of it happens on one machine. Planning a query over forty thousand partitions is single threaded coordinator work, and a coordinator that is pegged slows every query on the cluster regardless of how many workers are idle.
Stages, tasks, splits and drivers
These four words are used loosely in conversation and precisely in the engine, and mixing them up makes a profile unreadable.
A stage is a piece of the plan between two exchanges. A task is one stage running on one worker. A split is a unit of readable input, usually a file or a byte range within one. A driver is a thread of execution inside a task, processing one split at a time through a pipeline of operators.
So a leaf stage with ten thousand splits running on twenty workers with eight drivers each has a hundred and sixty drivers chewing through ten thousand splits. The split count divided by the driver count is roughly how many rounds of work the stage takes, which is the first number to compute when a scan is slow.
Exchanges are the only place data moves
Between stages, rows cross the network. There are three patterns and the plan names each one.
A gathering exchange collects everything onto one node, which is what a final aggregation or a top level sort does. A repartitioning exchange hashes rows by a key so that matching keys meet on the same node, which is what a distributed join and a grouped aggregation need. A replicating exchange sends every row to every node, which is what a broadcast join does to its small side.
Everything expensive in a distributed query is either a scan or an exchange. Operators between them are cheap by comparison, and a plan that looks slow is almost always slow at one of those two boundaries.
Memory lives in named pools
Workers do not have one pool of memory for everything, and the distinction decides which queries fail.
User memory is what a query's own operators hold: hash tables, sort buffers, aggregation state. System memory is what the engine holds on the query's behalf: exchange buffers, read buffers, output buffers. Limits are set separately on each, and the per query limit that teams tune is almost always the user limit.
A query that exceeds its user limit is killed with a clear message naming the query. A cluster that exhausts total memory behaves much worse, because the engine starts killing whichever queries it must to recover, and the victim is frequently not the cause.
Why a stateless engine is attractive, and what it costs
Having no storage means the engine can read anything with a connector, scale workers up and down freely, and never own a migration. Those are real advantages and they are why these engines spread.
The cost is that every query pays full price every time. There is no result cache that survives, no materialisation unless you build it yourself, no statistics unless the source provides them, and no control over layout. Every lesson about file sizing and clustering applies here with more force than it does in a warehouse, because the engine has no way to compensate.
So the engine is excellent at reaching data and indifferent at making repeated access cheap. Teams that understand that pair it with a lakehouse they maintain properly. Teams that do not use it as a warehouse replacement and then wonder why the same dashboard query costs the same every morning.
Reading the stage list before anything else
When a query is slow, the stage list is the first thing to open, and it answers the question most people try to answer by guessing.
Each stage reports the rows it produced, the bytes it moved, the wall time it took, the CPU time it consumed and the number of tasks it ran. The stage with most of the wall time is where the query lives, and comparing its wall time against its CPU time tells you whether it was computing or waiting.
A leaf stage with high wall time and low CPU time is waiting on storage, which means splits, file sizes or a slow source. A middle stage with high CPU and high rows is doing real work, which means the plan asked for too much data or the join order is wrong. A gathering stage with one task and a large row count is a plan mistake, because something that should have stayed distributed was collected onto one node.
Three readings, three different remedies, and the stage list distinguishes them in under a minute. Guessing from the total duration distinguishes nothing.
Operator counters worth knowing
Inside a stage, each operator reports its own input and output row counts, and the ratios between them are the useful signal.
A scan whose output rows are far below its input rows means a filter was pushed into the read, which is what you want. A scan whose output equals its input followed by a filter operator that drops most of them means the pushdown did not happen, and finding out why is usually the single largest win available.
A join whose output rows greatly exceed the sum of its inputs means the join keys are duplicated on both sides, which produces a multiplicative result that nobody intended. This is the most common cause of a query that ran for an hour and then failed on memory.
An aggregation whose input and output row counts are nearly equal means the grouping key is almost unique, so the aggregation is doing the work of a sort for no reduction in volume.
Where this engine is not distributed at all
One detail surprises people, and it is worth stating plainly because it changes how you write queries.
Some work has to happen on a single node. The final aggregation of a global count, the top level sort of an ordered result set, and the collection of output rows all funnel through one task. If a query orders a billion rows without a limit, one node sorts a billion rows, whatever the cluster size.
Window functions over a single partition behave the same way, because every row in the partition must be in one place to be ordered. A window specification with no partitioning clause over a large table is therefore a single node operation hidden inside distributed syntax.
The remedy is nearly always to reduce before gathering: aggregate first and sort the aggregate, add a limit so that the sort can be done partially on each node, or supply a partitioning clause so that the window splits across workers.