Handling Data Skew Adaptively In Spark Using Dynamic Repartitioning
19 likes10,315 views
This document discusses handling data skew in Spark using dynamic repartitioning. It presents an architecture for dynamic repartitioning that collects data characteristics during job execution, decides when repartitioning is needed based on those characteristics, and constructs a new hash function to redistribute the data more uniformly. It aims to mitigate the problem of slow tasks caused by skewed data distributions. Benchmark results show the approach can significantly reduce the size of the largest partition and execution time for jobs operating on skewed data.
More Related Content
What's hot (20)
Viewers also liked (20)
Similar to Handling Data Skew Adaptively In Spark Using Dynamic Repartitioning (20)
More from Spark Summit (20)
Recently uploaded (20)
Handling Data Skew Adaptively In Spark Using Dynamic Repartitioning
- 2. Introduction • Hungarian Academy of Sciences, Institute for Computer Science and Control (MTA SZTAKI) • Research institute with strong industry ties • Big Data projects using Spark, Flink, Cassandra, Hadoop etc. • Multiple telco use cases lately, with challenging data volume and distribution
- 3. Agenda • Our data-skew story • Problem definitions & aims • Dynamic Repartitioning • Architecture • Component breakdown • Repartitioning mechanism • Benchmark results • Visualization • Conclusion
- 4. Motivation • We have developed an application aggregating telco data that tested well on toy data • When deploying it against the real dataset the application seemed healthy • However it could become surprisingly slow or even crash • What did go wrong?
- 5. Our data-skew story • We have use-cases when map-side combine is not an option: groupBy, join • 80% of the traffic generated by 20% of the communication towers • Most of our data is 80–20 rule
- 6. The problem • Using default hashing is not going to distribute the data uniformly • Some unknown partition(s) to contain a lot of records on the reducer side • Slow tasks will appear • Data distribution is not known in advance • „Concept drifts” are common stage boundary & shuffle even partitioning skewed data slow task slow task
- 7. Aim Generally, make Spark a data-aware distributed data-processing framework • Collect information about the data-characteristics on-the-fly • Partition the data as uniformly as possible on-the-fly • Handle arbitrary data distributions • Mitigate the problem of slow tasks • Be lightweight and efficient • Should not require user-guidance spark.repartitioning = true
- 8. Architecture Slave Slave Slave Master approximate local key distribution & statistics 1 collect to master 2 global key distribution & statistics 3 4 redistribute new hash
- 9. Driver perspective • RepartitioningTrackerMaster part of SparkEnv • Listens to job & stage submissions • Holds a variety of repartitioning strategies for each job & stage • Decides when & how to (re)partition
- 10. Executor perspective • RepartitioningTrackerWorker part of SparkEnv • Duties: • Stores ScannerStrategies (Scanner included) received from the RTM • Instantiates and binds Scanners to TaskMetrics (where data- characteristics is collected) • Defines an interface for Scanners to send DataCharacteristics back to the RTM
- 11. Scalable sampling • Key-distributions are approximated with a strategy, that is • not sensitive to early or late concept drifts, • lightweight and efficient, • scalable by using a backoff strategy keys frequency counters keys frequency sampling rate of 𝑠" sampling rate of 𝑠" /𝑏 keys frequency sampling rate of 𝑠"%&/𝑏 truncateincrease by 𝑠" 𝑇𝐵
- 12. Complexity-sensitive sampling • Reducer run-time can highly correlate with the computational complexity of the values for a given key • Calculating object size is costly in JVM and in some cases shows little correlation to computational complexity (function on the next stage) • Solution: • If the user object is Weightable, use complexity-sensitive sampling • When increasing the counter for a specific value, consider its complexity
- 13. Scalable sampling in numbers • In Spark, it has been implemented with Accumulators • Not so efficient, but we wanted to implement it with minimal impact on the existing code • Optimized with micro-benchmarking • Main factors: • Sampling strategy - aggressiveness (initial sampling ratio, back-off factor, etc…) • Complexity of the current stage (mapper) • Current stage’s runtime: • When used throughout the execution of the whole stage it adds 5-15% to runtime • After repartitioning, we cut out the sampler’s code-path; in practice, it adds 0.2-1.5% to runtime
- 14. Main new task-level metrics • TaskMetrics • RepartitioningInfo – new hash function, repartition strategy • ShuffleReadMetrics • (Weightable)DataCharacteristics – used for testing the correctness of the repartitioning & for execution visualization tools • ShuffleWriteMetrics • (Weightable)DataCharacteristics – scanned periodically by Scanners based on ScannerStrategy • insertionTime • repartitioningTime • inMemory • ofSpills
- 15. Scanner • Instantiated for each task before the executor starts them • Different implementations: Throughput, Timed, Histogram • ScannerStrategy defines: • when to send to the RTM, • histogram-compaction level.
- 16. Decision to repartition • RepartitioningTrackerMaster can use different decision strategies: • number of local histograms needed, • global histogram’s distance from uniform distribution, • preferences in the construction of the new hash function.
- 17. Construction of the new hash function 𝑐𝑢𝑡, - single-key cut hash a key here with the probability proportional to the remaining space to reach 𝑙 𝑐𝑢𝑡. - uniform distribution top prominent keys hashed 𝑙 𝑐𝑢𝑡/ - probabilistic cut
- 18. New hash function in numbers • More complex than a hashCode • We need to evaluate it for every record • Micro-benchmark (for example String): • Number of partitions: 512 • HashPartitioner: AVG time to hash a record is 90.033 ns • KeyIsolatorPartitioner: AVG time to hash a record is 121.933 ns • In practice it adds negligible overhead, under 1%
- 19. Repartitioning • Reorganize previous naive hashing • Usually happens in-memory • In practice, adds additional 1-8% overhead (usually the lower end), based on: • complexity of the mapper, • length of the scanning process.
- 20. More numbers (groupBy) MusicTimeseries – groupBy on tags from a listenings-stream 1710 200 400 83M 134M 92M sizeof thebiggestpartition number of partitions over-partitioning overhead & blind scheduling 64% reduction observed 3% map-side overhead naive Dynamic Repartitioning 38% reduction
- 21. More numbers (join) • Joining tracks with tags • Tracks dataset is skewed • Number of partitions is set to 33 • Naive: • size of the biggest partition = 4.14M • reducer’s stage runtime = 251 seconds • Dynamic Repartitioning • size of the biggest partition = 2.01M • reducer’s stage runtime = 124 seconds • heavy map, only 0.9% overhead
- 22. Spark REST API • New metrics are available through the REST API • Added new queries to the REST API, for example: „what happened in the last 3 second?” • BlockFetches are collected to ShuffleReadMetrics
- 23. Execution visualization of Spark jobs
- 24. Repartitioning in the visualization heavy keys create heavy partitions (slow tasks) heavy is alone, size of the biggest partition is minimized
- 25. Conclusion • Our Dynamic Repartitioning can handle data skew dynamically, on- the-fly on any workload and arbitrary key-distributions • With very little overhead, data skew can be handled in a natural & general way • Visualizations can aid developers to better understand issues and bottlenecks of certain workloads • Making Spark data-aware pays off