dask / dask/dask-expr

Data Skew

Open
#417 1 comment 0 reactions 0 assignees View on GitHub
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

Open the contributing 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

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.