Data Skew
- Dominant language
- Python
- Stars
- 89
- Forks
- 26
- PR merge metrics
- No merged PRs in 30d
Description
I'm curious about mechanisms to address data skew in sorting/shuffling/merge operations, where some values are far more common than others. This comes up a lot when talking to Spark people as a common thing to watch out for. It's also a non-trivial pain point for some dask.dataframe users.
I think a useful benchmark to explore and become sensitive to these workloads is the RenRe workflow some folks here have.
### Quantiles help, but maybe not enough?
Maybe what we do today with pre-computing quantiles helps with some of this. It means that common values are more likely to be alone in their own partition, rather than with lots of other data (this would happen if we hash, for example). Maybe that's why it has been less of a concern for us than for Spark people historically?
Beyond that though, let's consider the case where there is a single value that is very common, and we're doing some join-like operation that expands it even further. How can we avoid memory blowing up?
### Solutions
Spark uses iterative algorithms, kind of like how we operate in dask.bag today. They consume chunks of dataframes and emit chunks of dataframes. We could maybe do something similar in some cases
After blockwise fusion could look at fused expressions and find any that have joins in them. Probably we could switch these to run the same set of blockwise operations, but in a for loop over smaller bits of data coming in. This is only beneficial if there are one of two other kinds of nodes in the fused expression after the join / expanding operation:
1. The beginning of an ApplyConcatApply sequence. This likely reduces the data, and then we can throw in an extra `concat` and `combine` at the end of the first `apply` stage.
2. Writing to parquet/csv where we probably write each chunk as a row group or something to some data sink
### Timing
This feels important, but not urgent. I don't think that we should think too much about it until after we've got dask-expr replacing dask.dataframe and good performance on benchmarks that don't include much skew.
Probably the one thing to do early-ish is to see how dask-expr and the new p2p system perform on the RenRe benchmark. cc @fjetter
Contributor guide
Research direction
No source file or test is named. Start with the RenRe workflow benchmark and compare dask-expr with the new p2p system; the early milestone is performance evidence, while a complete solution would require deciding how skewed join and expanding workloads should be handled.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- python
- Domain
- data-engineering, distributed-systems, performance
- Issue type
- Feature
- Difficulty
- 5/5
- Estimated time
- Over a week
- Activity status
- Stale
- Clarity
- Needs clarification
- Newbie friendliness
- 25/100