Optimizing PySpark jobs with skewed data distributions
Instruction: Discuss strategies to optimize a PySpark job that deals with highly skewed data, ensuring efficient processing and resource utilization.
Updated
Confirm skew from task-duration, input/shuffle-byte and spill distributions, then identify the relevant hot keys. A distributed groupBy(key).count() with a bounded top-key result is safer than collecting an unbounded RDD countByKey() dictionary to the driver. Also check whether skew comes from a join's many-to-many cardinality rather than just unequal key frequency.
For supported joins, inspect Adaptive Query Execution and its skew-join handling. It can split eligible skewed shuffle partitions, but does not repair every operator, hot-key aggregation or explosive join. A broadcast join may help when the build side safely fits each executor; verify its actual size and multiplicity.
Salting requires a semantics-preserving plan:
- For decomposable aggregates, distribute a hot key over several salt buckets, aggregate per key/salt, then merge per key. Merge sums and counts to compute a mean; averaging partial means weights unequal buckets incorrectly.
- For an equijoin, assign each hot-side row one salt and replicate the corresponding other-side rows over those salts, then join on key plus salt. Verify every original matching pair appears exactly once. Salting both sides independently can lose matches.
- Preserve null-key semantics, duplicate rows and any outer-join requirements. A recipe for an inner join is not automatically valid for an outer join.
Ordinary hashing or an RDD partitioner that maps each equal key to one partition cannot split that key. More partitions alone therefore do not solve a single-key bottleneck. Use a stable salt derived from a suitable row identity when repeatability matters; hashing a constant hot key provides no spread.
Compare repeated runs on representative data, checking outputs against an unsalted reference as well as tail latency, memory, shuffle and total work. Salting can add replication and an extra aggregation, so it should address a measured bottleneck rather than become a default transformation.