The lookup table was tiny. Why did the join move four terabytes?

The cheap option depends on which side moves

If the 20 MB product table fits comfortably in the usable memory budget of each worker, broadcast it to the workers already scanning the views. Each worker can join its local fact-table partitions against an in-memory copy of the product table. The 4 TB fact table does not need to be globally reshuffled just to find category labels.

Suppose the job has 80 workers and each gets one 20 MB copy. An illustrative total is about 1.6 GB of distributed payload, ignoring protocol overhead, retries and any repeated broadcasts. More importantly, per worker the build-side representation must fit with other operators and concurrent tasks. A 20 MB compressed file can occupy more memory after decoding and building a hash table.

The input still has to be scanned. Broadcasting does not make 4 TB vanish. It avoids redistributing that large fact side for this join. If the final aggregation groups by category, the partial counts may still need a smaller exchange later.

Apache Spark supports broadcast joins and several shuffle-based join strategies. A hint expresses a preference rather than a guarantee that a strategy is valid for every join type. The runtime plan, available statistics and size of the actual build side matter. Spark SQL performance tuning

Why did the engine shuffle four terabytes?

Start by reading the physical plan. Did the system choose a shuffle join because it thought the dimension table was much larger? Did the join happen before a selective filter? Was the small table padded into a wide intermediate relation containing columns not used by the query? Was broadcasting ruled out by the configured size threshold or join type? If an attempted broadcast ran out of memory or timed out, inspect the failure; do not assume Spark automatically retries every failed broadcast as a shuffle join.

The documented automatic broadcast threshold is 10 MiB by default in the linked Spark version, so a 20 MB lookup may already exceed it without any bad statistics. Check the actual configuration before blaming the planner. Statistics are especially important. A catalog might report that the dimension contains several billion rows from a previous version, while the filtered relation is actually tiny. Or a query might use a function on the join key, making partitioning information unusable. An optimizer could also make a reasonable choice based on an earlier stage whose cardinality estimates were wrong.

I would inspect estimated row counts alongside runtime row counts and sizes. Then I would check where projection and filters are applied. Removing twenty unnecessary columns before an exchange sometimes matters more than increasing network bandwidth.

If you must shuffle both sides, each input is repartitioned using a compatible function of the join key. Rows for one product ID are sent to a common destination. A worker can then join the two local partitions. That enables a distributed join but costs serialization, network transfer, potential spill, and possibly skew when some keys occur far more often than others.

A join can also have a severe skew problem even when the overall fact table is well distributed. If one product represents 30 percent of all views, all its rows may arrive at one join partition. A general increase in shuffle partitions does not change which partition owns that product key. The heavy-key alternatives now depend on whether the matching dimension row can be replicated and whether the join semantics allow a split.

Now the dimension grows to 80 GB

Broadcasting 80 GB to every worker would be a very different proposition. At 80 workers, the uncompressed aggregate payload is 6.4 TB before memory representation or network overhead. It may be impossible for an individual worker to hold the build side at all. Broadcasting is no longer the obvious choice.

A shuffle join may become appropriate. If both datasets are already partitioned on the same key with compatible conventions and trustworthy layout, some exchanges can be avoided. If a filtered subset of the 80 GB dimension is genuinely small, broadcasting that subset might still work. There is no timeless rule that says dimension tables always broadcast.

A physical plan should be justified with the actual post-filter sizes and memory per executor. Pinning a BROADCAST hint without checking those facts can convert a network problem into out-of-memory failures.

An interview answer should include the output contract

Suppose the dimension has two category rows for the same product ID. Joining can now duplicate a fact row even if the distributed strategy is otherwise correct. A faster join would simply produce a wrong count faster. Before measuring performance, confirm whether this is a one-to-one lookup, a versioned mapping, or intentionally one-to-many data.

For a versioned product catalog, an event-time category might be required rather than today's category. That affects the join condition and the data that must be retained. It is a business correctness question, not a shuffle optimization.

If I could ask the interviewer only one clarifying question, it would be: After all filters and projections, how large is the unique lookup relation per worker? That answer drives the next design decision far more directly than the raw size printed on the source table.

The team adopts the better join strategy and the daily job becomes fast. But analytics queries are still slow after the output reaches object storage. Our next problem is why the same bytes behave differently when broken into thousands of tiny files.