How Do You Handle Skewed Data in PySpark?
This is a critical PySpark interview question! Here’s the breakdown:
✅ What is Skewed Data?
A skewed partition in Spark occurs when the data is distributed unevenly across the available partitions, causing some partitions to be significantly larger than others. This imbalance creates performance bottlenecks because the tasks processing the large partitions, known as “straggler tasks,” take much longer to complete, while the tasks for smaller partitions finish quickly and remain idle.
Causes of Skewed Partitions
Skew typically arises during shuffle-intensive operations like joins or aggregations, where data with the same key must reside on the same executor.
Dominant Keys: The most common cause is when a few specific keys in a dataset appear much more frequently than others (e.g., a customer ID in an order table where one customer has 90% of the orders).
Poor Partitioning Strategy: Using a partitioning key that naturally has uneven data distribution (like a country column where 90% of records are from one country).
Data Quality Issues: High percentages of null or default values in a key column can cause all those records to end up in the same partition.
Consequences of data skewness in spark
- Slow Performance: The overall job performance is limited by the time it takes for the slowest straggler tasks to finish.
- Resource Inefficiency: Most executors sit idle, waiting for the few overloaded ones to complete their work.
- Out-of-Memory (OOM) Errors: The large partitions may exceed the memory capacity of a single executor, causing the job to fail or spill data to disk, which is much slower.
Detection of data skewness in spark
Skew can be detected using the Spark UI by observing:
- Uneven Task Durations: Some tasks take significantly longer (minutes vs. seconds) within the same stage.
- Large Shuffle Read Sizes: A few tasks show disproportionately larger “Shuffle Read Size” metrics compared to the median.
How to Handle Skewed Data:
Following are the few techniques used to handle the skewed data
1. Salting: Add a random prefix to keys to distribute data more evenly.
2. **Custom Partitioning**: Manually repartition the data based on a more balanced key or use a custom partitioner to ensure even distribution.
3. Broadcast Joins: If one of the datasets is small enough (under 10MB or a configurable threshold), broadcasting it to all executors avoids the shuffle operation entirely.
4. Repartitioning: Use 𝚛𝚎𝚙𝚊𝚛𝚝𝚒𝚝𝚒𝚘𝚗() to increase the number of partitions and reduce skew.
5. Adaptive Query Execution (AQE): Spark 3.x and later can dynamically handle skewed joins by splitting large partitions into smaller ones at runtime.
Example (Salting):
𝚏𝚛𝚘𝚖 𝚙𝚢𝚜𝚙𝚊𝚛𝚔.𝚜𝚚𝚕.𝚏𝚞𝚗𝚌𝚝𝚒𝚘𝚗𝚜 𝚒𝚖𝚙𝚘𝚛𝚝 𝚌𝚘𝚕, 𝚌𝚘𝚗𝚌𝚊𝚝, 𝚕𝚒𝚝, 𝚛𝚊𝚗𝚍
# 𝙰𝚍𝚍 𝚊 𝚛𝚊𝚗𝚍𝚘𝚖 𝚜𝚊𝚕𝚝 𝚝𝚘 𝚝𝚑𝚎 𝚔𝚎𝚢
𝚍𝚏_𝚜𝚊𝚕𝚝𝚎𝚍 = 𝚍𝚏.𝚠𝚒𝚝𝚑𝙲𝚘𝚕𝚞𝚖𝚗(“𝚜𝚊𝚕𝚝𝚎𝚍_𝚔𝚎𝚢”, 𝚌𝚘𝚗𝚌𝚊𝚝(𝚌𝚘𝚕(“𝚔𝚎𝚢”), 𝚕𝚒𝚝(“_”), (𝚛𝚊𝚗𝚍() * 𝟷𝟶).𝚌𝚊𝚜𝚝(“𝚒𝚗𝚝”)))
# 𝙿𝚎𝚛𝚏𝚘𝚛𝚖 𝚘𝚙𝚎𝚛𝚊𝚝𝚒𝚘𝚗𝚜 𝚘𝚗 𝚝𝚑𝚎 𝚜𝚊𝚕𝚝𝚎𝚍 𝚔𝚎𝚢
𝚛𝚎𝚜𝚞𝚕𝚝 = 𝚍𝚏_𝚜𝚊𝚕𝚝𝚎𝚍.𝚐𝚛𝚘𝚞𝚙𝙱𝚢(“𝚜𝚊𝚕𝚝𝚎𝚍_𝚔𝚎𝚢”).𝚌𝚘𝚞𝚗𝚝()
Pro Tip: Monitor your Spark UI to identify skewed partitions and apply the right strategy!
🔗 See also: Alternative Approaches to Handling Skewed Data in PySpark (Part 2)