Distributed Data Processing with Spark and Hadoop Questions

Direct answer

Spark chooses among three main join strategies: BROADCAST-HASH-JOIN (ship the smaller side to every executor, no shuffle of the larger side), SHUFFLE-HASH-JOIN (shuffle both sides by key, build an in-memory hash table on the smaller SHUFFLED side per partition), and SORT-MERGE-JOIN (shuffle both sides by key, sort each partition, merge; Spark's general-purpose default for two large sides). For a 200M-row events table joined against a 50K-row countries lookup, Spark will almost certainly choose BROADCAST-HASH-JOIN automatically, since a 50K-row lookup table (even at a generous row width) is very plausibly well under the spark.sql.autoBroadcastJoinThreshold default (10 MB), making this one of the cases where the DEFAULT threshold correctly and automatically catches a genuinely safe broadcast without needing an explicit hint.

Structured elaboration

Broadcast-hash-join. Ships the SMALLER side's data, once per executor, so the larger side never shuffles; the smaller side is built into an in-memory HASH TABLE on each executor, and the larger side's local partition rows probe that hash table directly. Chosen AUTOMATICALLY when a side's estimated size is under spark.sql.autoBroadcastJoinThreshold, or explicitly via F.broadcast()/a SQL hint (the mechanics and threshold-override discipline in depth).

Shuffle-hash-join. BOTH sides are shuffled by the join key (unlike a broadcast join, which shuffles neither side), then, WITHIN each resulting partition, a hash table is built from the SMALLER side's shuffled data for that partition, and the larger side's rows in that same partition probe it. This is Spark's LEAST commonly chosen strategy in practice: it requires the smaller (post-shuffle, per-partition) side to still fit comfortably in memory for the hash-table build, and Spark's planner generally prefers either a broadcast join (if genuinely safe) or a sort-merge join (the more robust general-purpose default) over shuffle-hash-join unless specifically hinted (/*+ SHUFFLE_HASH(table) */) or under specific conditions where it is estimated favorable.

Sort-merge-join. BOTH sides shuffled by the join key, then EACH partition's data is SORTED by the key, and matching rows are found via a merge-scan (an efficient linear pass, since both sides are now sorted). This is Spark's general-purpose DEFAULT for a join between two sides where NEITHER is small enough to broadcast; it is more memory-robust than shuffle-hash-join (a sort-based merge does not need to hold a full hash table in memory) at the cost of the SORT step's own overhead ($O(n \log n)$ per partition).

Which strategy Spark will likely pick for 200M rows vs 50K rows, and why. A 50K-row lookup table, even at a generous width (say 200 bytes/row, giving roughly $50{,}000 \times 200 = 10$ MB), sits right AT or comfortably UNDER the DEFAULT 10 MB auto-broadcast threshold for a typical lookup table's row width (most countries-style reference tables are narrow: a code, a name, maybe a region, well under 200 bytes/row in practice); this is a case where the CONSERVATIVE default threshold correctly catches a genuinely safe broadcast automatically, unlike the LARGER 2-million-row dimension example, which typically DOES need an explicit override. Spark's cost-based optimizer, seeing countries' estimated size well under the threshold, chooses BroadcastHashJoin WITHOUT needing any hint at all for this specific scale.

How to inspect and influence the planner. INSPECT: df.explain() (or .explain("formatted") for a more structured, readable breakdown) shows the CHOSEN physical plan directly (BroadcastHashJoin vs SortMergeJoin vs ShuffleHashJoin appears explicitly in the plan's operator names). INFLUENCE: F.broadcast(df) (DataFrame API) or /*+ BROADCAST(table) */ / /*+ SHUFFLE_HASH(table) */ / /*+ MERGE(table) */ (SQL hints) explicitly request a specific strategy, overriding the planner's own cost-based estimate; spark.sql.autoBroadcastJoinThreshold adjusts the GLOBAL automatic-broadcast size cutoff (a blunter, session-wide lever versus a per-join hint).

Worked example

events.join(countries, "country_code"), events at 200M rows, countries at 50K rows with a typical narrow row width (country_code, country_name, region, roughly 60-80 bytes/row), giving an estimated table size of roughly $50{,}000 \times 70 \approx 3.5$ MB, comfortably under the 10 MB default threshold.

Expected physical plan: BroadcastHashJoin, chosen automatically, with countries as the broadcast side; df.explain() against this join shows BroadcastHashJoin explicitly in the plan, the concrete way to CONFIRM this expectation rather than assume it.

If countries instead had an unusually WIDE row (many additional text columns, pushing estimated size to, say, 50 MB, over the default threshold): the automatic decision would flip to SortMergeJoin even though 50K rows still SOUNDS small; an explicit F.broadcast(countries) hint, validated against actual executor memory, would restore the broadcast join deliberately, exactly the "the default threshold can miss a genuinely safe LARGER broadcast" gap with its own worked 2M-row example.

Trade-offs and pitfalls

• Common mistake: assuming a "small" table (as measured by ROW COUNT alone, 50K rows sounding trivially small) is automatically safe to broadcast without checking its actual estimated BYTE size; row count and byte size can diverge significantly depending on row width, and the threshold Spark actually compares against is a byte-size estimate, not a row count.
• Common mistake: never checking explain() at all and simply assuming the "obviously right" strategy was chosen; for THIS specific scenario the automatic choice is very likely correct, but relying on that assumption without verification is a habit that fails silently the moment table sizes shift (a countries table that grows wider over time, as in the trade-off above).
• Shuffle-hash-join is worth knowing exists, but is rarely the RIGHT strategy to explicitly force; sort-merge-join's greater memory robustness (no full in-memory hash table required) generally makes it the safer general-purpose choice when neither side is broadcast-safe, and Spark's own planner reflects this preference by default.
• The EXPLAIN output's plan can change between Spark VERSIONS or with AQE enabled (runtime join re-planning directly); confirming the plan for a SPECIFIC Spark version and configuration, rather than assuming behavior carries over unchanged from a different version or a different AQE setting, is the more reliable habit.

Direct answer

For typical feature-engineering and ETL work, prefer DataFrames (or, in Scala, Datasets when compile-time type safety is valuable): they get Catalyst/Tungsten's optimizations (predicate pushdown, whole-stage codegen, off-heap memory management) that RDDs bypass entirely, since RDD transformations are opaque black-box functions to the optimizer. RDDs remain the right choice specifically when working with genuinely unstructured data that does not fit a schema, when needing FINE-GRAINED control over physical partitioning/execution that the DataFrame API does not expose, or when implementing an algorithm whose logic is fundamentally imperative/iterative in a way that resists a clean declarative expression.

Structured elaboration

Performance (Catalyst/Tungsten). DataFrame/Dataset operations compile through Catalyst's full optimization pipeline (the mechanism in depth): predicate/projection pushdown, join reordering, whole-stage code generation, off-heap Tungsten binary memory management. RDD transformations are arbitrary compiled JVM/Python functions that Catalyst cannot see INSIDE (the same opacity issue for UDFs), so NONE of these optimizations apply to hand-written RDD map/filter/reduceByKey chains; for equivalent LOGIC, a DataFrame-API implementation is generally faster, often substantially, purely from these optimizations, without the developer needing to hand-tune anything.

Type safety. RDDs are typed (RDD[T]) but their transformations are checked only insofar as the compiler checks the closure's own code, not against any SCHEMA; a DataFrame (untyped, Row-based in both Scala and Python) has NO compile-time column-name/type checking at all, errors surface only at RUNTIME when a referenced column does not exist or has an unexpected type. Scala's Dataset[T] bridges this gap: a genuinely typed, compile-time-checked API (Dataset[Person] where Person is a case class) that STILL gets Catalyst/Tungsten's optimizations (unlike a raw RDD), the closest thing to "the best of both worlds" for Scala specifically; Python has no equivalent typed-Dataset construct (Python's own dynamic typing makes a Scala-style compile-time-checked Dataset structurally not applicable the same way), so this specific trade-off is Scala-only.

When to still choose RDDs. (1) Genuinely unstructured/non-tabular data: arbitrary binary blobs, or data whose "rows" do not have a stable, useful schema (some scientific/graph-processing workloads), where forcing a DataFrame schema adds friction without benefit. (2) Fine-grained control needs: a custom Partitioner, explicit control over WHEN data is shuffled/cached/checkpointed at a granularity the DataFrame API's more automatic, Catalyst-driven decisions do not directly expose. (3) Fundamentally imperative/iterative algorithms: some graph or numerical algorithms (a custom iterative optimization loop with complex, stateful, non-tabular per-iteration logic) map more naturally onto RDD-style explicit transformations than onto a declarative DataFrame query, though this case has narrowed considerably as the DataFrame/Dataset API's expressiveness (window functions, complex nested types, flatMapGroupsWithState-style stateful processing) has matured.

Worked example

A feature-engineering pipeline: read raw events, join against a dimension table, compute several derived numeric features, write the result.

As RDDs (not recommended for this shape):

events_rdd = sc.textFile("s3://bucket/events/").map(parse_line)
dim_rdd = sc.textFile("s3://bucket/dims/").map(parse_line).map(lambda r: (r.id, r))
joined = events_rdd.map(lambda r: (r.dim_id, r)).join(dim_rdd)
features = joined.map(compute_features)

Every step here (parse_line, compute_features) is an opaque Python function; Catalyst cannot push a filter into the text read, cannot prune unused columns (there is no COLUMN concept at the RDD level at all, just whatever structure parse_line happens to produce), and the join gets none of the broadcast/bucketing-aware planning for DataFrame joins.

As DataFrames (recommended):

events_df = spark.read.schema(events_schema).parquet("s3://bucket/events/")
dim_df = spark.read.schema(dim_schema).parquet("s3://bucket/dims/")
features_df = (events_df
    .join(F.broadcast(dim_df), "dim_id")
    .withColumn("feature_a", F.col("amount") / F.col("baseline"))
    .withColumn("feature_b", F.log1p(F.col("count"))))

Catalyst sees the FULL chain (schema-aware scan, join, two derived-column computations) at once, can push down any applicable filters, prune unused columns from the Parquet scan, and choose the broadcast join, with the whole compiled chain executed via Tungsten's whole-stage codegen; for equivalent logic, this DataFrame version is the strictly better default, and throughout (window functions, joins, aggregations) consistently use the DataFrame API for exactly this reason.

Trade-offs and pitfalls

• Common mistake: defaulting to RDDs out of familiarity (a common pattern for engineers who learned Spark's older API first) for tabular, schema-friendly data that would benefit substantially from the DataFrame API's optimizations; the RDD API remains fully supported but should be a DELIBERATE choice for one of the specific reasons above, not a default.
• Common mistake: assuming Dataset[T]'s type safety comes for free performance-wise; a Dataset that frequently converts to/from RDD-style typed operations (.map with an arbitrary Scala function, rather than native column expressions) can LOSE some of Catalyst's optimization visibility for those specific operations, similar to how a UDF is opaque to Catalyst even within an otherwise DataFrame/Dataset pipeline; type safety and full optimizer visibility are not automatically the same thing, worth checking the actual physical plan rather than assuming.
• Dataset[T]'s value for JVM library interoperability: beyond compile-time safety, Dataset[T] lets Scala/Java code pass STRONGLY-TYPED domain objects (not generic Rows) directly into and out of Spark pipelines, valuable when a pipeline needs to interoperate with existing JVM libraries/domain models that expect specific typed objects rather than Spark's generic Row representation, a practical software-engineering benefit distinct from the pure performance question.
• The RDD API is not deprecated or going away; for the specific, narrower cases named above (genuinely unstructured data, fine-grained partitioning control, fundamentally imperative algorithms), it remains the right tool, and the guidance here is about the DEFAULT choice for the COMMON case (feature engineering, ETL, this question's own framing), not a blanket claim that RDDs are obsolete.

Direct answer

Partition pruning skips entire directories (partitions) whose partition-column value cannot satisfy a query's filter, without opening a single file inside them; predicate pushdown goes further WITHIN the files that ARE read, using Parquet's own per-row-group statistics (min/max values, and null counts) to skip entire ROW GROUPS that cannot match the filter, without decompressing or scanning their actual row data. The common reasons pruning fails differ by WHICH mechanism is involved, and are worth naming precisely rather than lumped together: predicate pushdown is reliably defeated by wrapping the filtered column in a function (confirmed directly below), while partition pruning is more resilient to that specific pattern than commonly assumed, since it evaluates predicates against the already-enumerated partition VALUES rather than against file statistics; genuinely non-deterministic expressions defeat both.

Structured elaboration

Partition pruning mechanics. A table physically laid out as s3://bucket/orders/order_date=2026-07-01/, order_date=2026-07-02/, etc. (Hive-style partition directories) lets Spark's planner read the PARTITION VALUES directly from directory NAMES (no file content needs to be touched at all), decode them into typed values, and evaluate a query's filter EXPRESSION against those already-decoded values in memory, listing only the directories whose value satisfies it; a query filtering WHERE order_date = '2026-07-15' against 2 years of daily partitions (roughly 730 directories) reads exactly 1, entirely from directory-name comparison.

Predicate pushdown mechanics. WITHIN the (already partition-pruned) set of files that must be opened, columnar formats like Parquet store PER-ROW-GROUP statistics (min/max value, and null count, for each column) in the file's own footer metadata; for pushdown to engage, Spark's planner must be able to express the filter as one of a SPECIFIC, LIMITED set of pushable filter shapes (equality, range comparisons, IN-lists, IS NULL, directly against a raw column) that the Parquet reader's stats-based filter understands; when it can, an entire row group is skipped ENTIRELY (no decompression, no row-by-row scan) whenever the filter's value falls outside that row group's min/max range.

Common reasons pruning fails, MEASURED rather than assumed. I initially drafted this section from a simplified mental model (that wrapping either a partition or a pushdown column in a function defeats pruning identically) and did not ship it without checking; running both cases against a real partitioned Parquet table showed the two mechanisms behave DIFFERENTLY under the identical pattern:

• Predicate pushdown IS defeated by a function wrapped around the filtered column, confirmed directly (worked example below): WHERE ROUND(amount, 0) > 150 produces an EMPTY PushedFilters list in the physical plan (the wrapped expression cannot be expressed as a pushable filter shape), falling back to evaluating it as a DataFilter after decompressing and reading every row, while the direct WHERE amount > 150 shows PushedFilters: [GreaterThan(amount,150.0)] engaged.
• Partition pruning is MORE RESILIENT to a function wrap than commonly assumed, also confirmed directly: WHERE dayofmonth(order_date) = 3 STILL appears inside PartitionFilters in the physical plan and correctly prunes to only the matching directories, because the planner evaluates the wrapped expression against each partition's small, already-enumerated set of decoded values (cheap, since there are typically far fewer distinct partitions than rows), rather than needing to invert the function symbolically the way row-group-statistics-based pushdown would.
• Genuinely non-deterministic expressions defeat BOTH mechanisms. WHERE order_date = current_date() - 1 depends on WHEN the query runs, not a fixed value the planner can resolve once and compare against either partition directory names or row-group statistics; some Spark versions evaluate a non-deterministic expression once upfront and treat the result as effectively constant for that execution, but relying on that behavior rather than computing the concrete bound explicitly in application code is fragile across versions.

Worked example

from pyspark.sql import SparkSession
from pyspark.sql import functions as F
import tempfile, os

spark = SparkSession.builder.master("local[2]").appName("s90").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")

tmpdir = tempfile.mkdtemp()
df = spark.createDataFrame(
    [(f"2026-07-{d:02d}", i, float(i * 10)) for d in range(1, 6) for i in range(1, 21)],
    ["order_date", "customer_id", "amount"])
df.write.partitionBy("order_date").mode("overwrite").parquet(tmpdir)
read_back = spark.read.parquet(tmpdir)

# PARTITION pruning: direct vs. function-wrapped partition column.
part_direct = read_back.filter(F.col("order_date") == "2026-07-03")
part_wrapped = read_back.filter(F.dayofmonth(F.to_date("order_date")) == 3)

# PREDICATE PUSHDOWN: direct vs. function-wrapped non-partition column.
push_direct = read_back.filter(F.col("amount") > 150)
push_wrapped = read_back.filter(F.round(F.col("amount"), 0) > 150)

def plan_line(df, needle):
    plan = df._jdf.queryExecution().simpleString()
    return next(l.strip() for l in plan.split("\n") if needle in l)

print("Partition, direct   :", plan_line(part_direct, "PartitionFilters"))
print("Partition, wrapped  :", plan_line(part_wrapped, "PartitionFilters"))
print("Pushdown,  direct   :", plan_line(push_direct, "PushedFilters"))
print("Pushdown,  wrapped  :", plan_line(push_wrapped, "PushedFilters"))
print("\nAll four correctness-equal:",
      part_direct.count() == 20, part_wrapped.count() == 20,
      push_direct.count() == push_wrapped.count())

os.system(f"rm -rf {tmpdir}")
spark.stop()

Output (actually executed with python3.12 + pyspark 3.5.1, Java 17, local[2]; lines trimmed to the relevant clause):

Partition, direct   : ...PartitionFilters: [isnotnull(order_date#12), (order_date#12 = 2026-07-03)]...
Partition, wrapped  : ...PartitionFilters: [isnotnull(order_date#12), (dayofmonth(order_date#12) = 3)]...
Pushdown,  direct   : ...PushedFilters: [IsNotNull(amount), GreaterThan(amount,150.0)]...
Pushdown,  wrapped  : ...PushedFilters: []...

All four correctness-equal: True True True

The function-wrapped PARTITION filter still appears inside PartitionFilters and prunes correctly (20 rows either way), while the function-wrapped PUSHDOWN filter produces an EMPTY PushedFilters: [], confirmed defeated. All four queries return the CORRECT result regardless (pruning is a performance optimization, never a correctness requirement: a query that fails to prune still computes the right answer, just slower), which is exactly why this specific failure mode is invisible without checking the physical plan directly.

Trade-offs and pitfalls

• Common mistake, and the one this answer's own first draft nearly made: assuming partition pruning and predicate pushdown fail under IDENTICAL conditions because both are commonly described together as "pruning"; they are two DIFFERENT mechanisms (directory-enumeration-based versus file-statistics-based) with different resilience to a function-wrapped column, and treating them as interchangeable risks either an incorrect diagnosis in production or, as here, an incorrect claim in a design document.
• Common mistake: wrapping a NON-partition filtered column in a function out of habit (WHERE ROUND(amount, 0) > 150 instead of a direct comparison) without realizing it silently defeats predicate pushdown specifically; since the query still returns the CORRECT result either way, this defect is invisible without explicitly checking PushedFilters in the physical plan.
• Assuming explain() output is self-evidently clear about whether pruning engaged is itself a common mistake; the reliable signal is specifically the CONTENT of PartitionFilters (for partition pruning) and PushedFilters (for predicate pushdown) in the physical plan, worth learning to read those two specific sections rather than skimming the plan generally or trusting a mental model not checked against the actual plan.
• Predicate pushdown's row-group-skipping benefit depends on the DATA actually being somewhat SORTED or clustered by the filtered column WITHIN each file; a file whose rows are in random order relative to amount has row groups whose min/max ranges all overlap heavily, so few or no row groups end up skippable even when pushdown is technically engaged, a data-layout concern distinct from whether pushdown itself fired.

Direct answer

A NARROW transformation's output partition depends on data from only ONE parent partition (map, filter, union), so it can execute entirely within the SAME partition with no data movement across the network; a WIDE transformation's output partition can depend on data from MULTIPLE parent partitions (groupByKey, reduceByKey, join), which requires a SHUFFLE, physically moving data across the network so all values for a given key land on the same partition. A shuffle matters for performance because it is Spark's single most expensive operation class: it involves disk I/O (writing shuffle files), network I/O (transferring them between executors), and serialization overhead, categorically more costly than a narrow transformation's purely local, in-memory work.

Structured elaboration

The formal distinction. A transformation is narrow if EVERY output partition can be computed from AT MOST one input partition (Spark can determine this from the transformation's own definition, without looking at the actual data); map, filter, mapPartitions, and union (each output partition traces to exactly one, or in union's case one of several already-aligned, input partitions) are narrow. A transformation is wide if an output partition may need data from MULTIPLE input partitions, because the operation needs to bring together all values sharing the same KEY, and those values could originally live on ANY of the source partitions; groupByKey, reduceByKey, join, distinct, and repartition are wide, since satisfying "all values for key K in one place" requires a shuffle REGARDLESS of how the source data happened to be laid out.

Why the shuffle boundary matters for performance. A narrow transformation's tasks run entirely INDEPENDENTLY and in parallel with no cross-task coordination needed, since each task only needs its own partition's data; Spark can PIPELINE a chain of narrow transformations into a single stage with no materialization in between (exactly this stage-boundary mechanism). A wide transformation forces a STAGE BOUNDARY: all of the PRODUCING stage's tasks must write their shuffle output, and the CONSUMING stage's tasks cannot start reading until that shuffle output exists, introducing a synchronization point (the whole cluster effectively waits for the slowest producer task before the shuffle read can complete) plus the disk/network/serialization cost of moving the data itself; this combination is why a job's wide transformations, not its narrow ones, typically dominate total runtime and are the first place to look when optimizing.

Precisely when a wide transformation triggers a shuffle. The shuffle is triggered by the NEED to co-locate data by key across the cluster, not by the operation's name alone: reduceByKey still shuffles, but its MAP-SIDE COMBINE (partially reducing values for a key on each source partition BEFORE the shuffle) moves less data across the network than groupByKey's shuffle of every raw value; both are wide (both require a shuffle to complete), but they are not EQUALLY expensive shuffles for the same logical result, illustrating that "wide" is about WHETHER a shuffle happens, and the specific operation chosen still matters for HOW MUCH data that shuffle moves.

Refactoring to reduce wide transformations. Prefer reduceByKey/aggregateByKey (or the DataFrame API's groupBy().agg(), which gets the equivalent map-side-combine optimization automatically via Catalyst) over groupByKey whenever the aggregation is COMBINABLE (this exact fix in depth). Filter/project data DOWN (narrow transformations) BEFORE a wide transformation, not after, so the shuffle moves less data (a filter before a join, rather than joining everything and filtering the joined result, is a common, easy, purely-reordering win). If the SAME wide transformation's shuffle result is needed by multiple downstream branches, cache it rather than letting each branch trigger its own re-shuffle from the same lineage (exactly this pattern with executed output).

Worked example

from pyspark.sql import SparkSession
from pyspark.sql import functions as F

spark = SparkSession.builder.master("local[2]").appName("s88").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")

events = spark.createDataFrame(
    [("u1", "click"), ("u1", "view"), ("u2", "click"), ("u1", "click"), ("u3", "view")],
    ["user_id", "event_type"])

# NARROW: filter -- each output partition depends on exactly one input
# partition; no shuffle needed, executes entirely locally.
clicks_only = events.filter(F.col("event_type") == "click")

# WIDE: groupBy triggers a shuffle -- rows for the SAME user_id must be
# co-located, and any user's rows could originally be on any partition.
counts = clicks_only.groupBy("user_id").agg(F.count("*").alias("click_count"))

# Confirm via the physical plan: filter has NO Exchange, groupBy DOES.
filter_plan = clicks_only._jdf.queryExecution().simpleString()
groupby_plan = counts._jdf.queryExecution().simpleString()
print("filter() plan contains 'Exchange' (shuffle) -- expect False:", "Exchange" in filter_plan)
print("groupBy() plan contains 'Exchange' (shuffle) -- expect True:", "Exchange" in groupby_plan)
counts.orderBy("user_id").show()

Output (actually executed with python3.12 + pyspark 3.5.1, Java 17, local[2]):

filter() plan contains 'Exchange' (shuffle) -- expect False: False
groupBy() plan contains 'Exchange' (shuffle) -- expect True: True
+-------+-----------+
|user_id|click_count|
+-------+-----------+
|     u1|          2|
|     u2|          1|
+-------+-----------+

Directly inspecting the physical plan confirms filter (narrow) produced NO Exchange operator while groupBy (wide) DID, hard evidence of the shuffle boundary rather than an assumption from the transformation's name alone.

Trade-offs and pitfalls

• Common mistake: assuming any operation whose NAME sounds like it "combines" data must be wide, or vice versa; union (combining two datasets) is narrow (each output partition maps to exactly one of the two inputs' corresponding partitions, no key-based co-location needed), while a single-column distinct (which sounds like a simple filter) is WIDE, since it needs to co-locate identical values to detect duplicates across the whole dataset.
• Common mistake: reordering a filter to AFTER a join/groupBy instead of before, shuffling data that will just be discarded afterward; pushing selective filters as early as possible, before any wide transformation, is one of the highest-leverage, lowest-risk optimizations available (and one Catalyst's own predicate pushdown will do automatically for many DataFrame-API cases, though verifying via explain() rather than assuming is worth the habit).
• Reducing the NUMBER of wide transformations is not the same goal as reducing the SHUFFLE VOLUME of the wide transformations that remain; reduceByKey versus groupByKey for the SAME aggregation does not remove a shuffle (both still shuffle), it reduces how MUCH data that unavoidable shuffle moves, a distinct, complementary optimization axis from simply having fewer wide operations in the pipeline.
• Lineage and execution planning depend directly on this narrow/wide distinction: it is precisely WHERE a wide transformation sits in the DAG that determines STAGE boundaries, and it is precisely the SHUFFLE boundary that determines where Spark can and cannot simply RECOMPUTE lost data from lineage without needing a broader stage re-execution, tying this question's own distinction directly to both execution planning and fault-tolerance mechanics.

Direct answer

Join skew shows up in the Spark UI as a small number of tasks in the join's shuffle stage with dramatically longer duration and dramatically larger shuffle-read size than their siblings, the same straggler-task signature, but specifically localized to a JOIN stage rather than a pure aggregation. To quantify skew BEFORE running the full join, sample both sides' join-key distributions cheaply (a groupBy(key).count() on a fraction of the data, or, more precisely, on the FULL key column alone without the expensive join itself) and compute a concentration statistic (the top key's share of total rows) to decide whether mitigation is needed before committing to the full, expensive run.

Structured elaboration

Spark UI signs of join skew specifically. In the join stage's task list: a small number of tasks with shuffle-read size (and correspondingly, duration) far above the median, the general skew signature; specific to a JOIN (versus a pure aggregation), also check whether the skew is coming from ONE side or BOTH sides of the join (a task fetching a disproportionate amount from the LEFT side's shuffle output versus the RIGHT side's can indicate which side's key distribution is actually the problem, informing which side needs salting or whether a broadcast of the OTHER, non-skewed side is the simpler fix); and disk spill on exactly the same slow tasks, confirming the oversized partition genuinely exceeded execution memory, not just took longer for an unrelated reason.

Job logs. Executor logs for the specific slow tasks may show explicit GC-time elevation (large partitions mean more live objects during the join's build/probe phase) or shuffle-fetch warnings; cross-referencing which HOST ran the slow tasks also rules out a node-level infrastructure cause before concluding the join key itself is skewed.

Programmatic sampling to quantify skew before running the full job. The key insight: computing the KEY DISTRIBUTION is far cheaper than running the actual join, since it needs only ONE side's key column (not both sides, not the full row width, not the join computation itself).

from pyspark.sql import functions as F

def quantify_skew(df, key_col, top_n=10):
    total = df.count()
    key_counts = (df.select(key_col).groupBy(key_col).count()
                    .orderBy(F.desc("count"))
                    .limit(top_n))
    rows = key_counts.collect()
    for r in rows:
        share = r["count"] / total
        print(f"{r[key_col]}: {r['count']} rows, {share:.1%} of total")
    return rows[0]["count"] / total if rows else 0.0

Running quantify_skew(large_side_df, "customer_id") before attempting the full join gives a concrete, cheap-to-compute number (the top key's SHARE of total rows) that can be checked against a threshold (threshold design in more depth) BEFORE committing to the full join's cost; a 40% top-key share is a strong, actionable signal to salt or otherwise mitigate proactively, while a 2% top-key share suggests the join is likely fine as-is and does not need proactive intervention.

Worked example

I ran the quantification approach against a synthetic dataset with the SAME 95%-skewed key used ("HOT" accounting for 950 of 1,000 rows, spread across 6 distinct keys):

quantify_skew(large_df, "key")

Output (actually executed with python3.12 + pyspark 3.5.1 + Java 17, local[2], rebuilding the same 1,000-row/95%-skewed dataset shape described above):

HOT: 950 rows, 95.0% of total
k1: 10 rows, 1.0% of total
k2: 10 rows, 1.0% of total
k3: 10 rows, 1.0% of total
k4: 10 rows, 1.0% of total
k5: 10 rows, 1.0% of total

A 95% top-key share is an extreme, unambiguous signal to mitigate BEFORE running the full join against this key; this is exactly the quantification step that should precede committing to the salting implementation, closing the loop between "detect proactively" (this question) and "mitigate" (the salting question) as two connected steps of one workflow, not two unrelated topics.

Trade-offs and pitfalls

• Common mistake: running df.groupBy(key).count() against the FULL row width (selecting all columns) instead of JUST the key column before aggregating; selecting only the key column first (df.select(key_col).groupBy(key_col).count()) lets Catalyst's projection pruning (the mechanism) avoid reading or shuffling any of the other, unneeded columns, making the sampling check itself meaningfully cheaper than a naive full-row count-by-key.
• Common mistake: sampling too small a FRACTION of the data when the skew is itself RARE-BUT-SEVERE (a key that is uncommon in a small sample but genuinely dominant at full volume); for a key distribution suspected to have this shape, computing the count over the FULL key column (cheap, since it excludes the other columns and the join itself) is more reliable than sampling a fraction, and is usually still far cheaper than running the actual join.
• A top-key-share threshold alone does not distinguish "one moderately hot key" from "many moderately warm keys collectively causing spread-out but real skew"; for the fullest picture, computing the coefficient of variation across ALL key counts (not just the single top key's share) gives a more complete signal than the top-1 share alone, though the top-1 share is a fast, useful FIRST check.
• Diagnosing which SIDE of a join is skewed matters for choosing the right mitigation: if only the large side has a skewed key, salting the large side plus replicating the small side is appropriate; if the "small" side turns out to itself have significant duplication on the join key (a data-quality issue, not the volume asymmetry salting is designed for), the real fix may be upstream deduplication rather than a join-strategy change at all.