[SPARK-49640][PS] Apply reservoir sampling in SampledPlotBase
#48105
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
zhengruifeng
commented
Sep 13, 2024
| F.monotonically_increasing_id().alias(id_col_name), | ||
| ) | ||
| .sort(rand_col_name) | ||
| .limit(max_rows + 1) |
There was a problem hiding this comment.
sort + limit is likely be optimized to TakeOrderedAndProject which output single partition, this coalesce here is just used to guarantee the partitioning.
zhengruifeng
commented
Sep 13, 2024
| .sort(rand_col_name) | ||
| .limit(max_rows + 1) | ||
| .coalesce(1) | ||
| .sortWithinPartitions(id_col_name) |
There was a problem hiding this comment.
using local sorting to avoid unnecessary shuffle
|
will send separate PR for the new dataframe plotting |
zhengruifeng
changed the title
SampledPlotBaseSampledPlotBase
xinrong-meng
approved these changes
Sep 17, 2024
|
I’m wondering if this is considered a “user-facing change” since the sampling directly affects the appearance of the plot |
|
Thank you @zhengruifeng ! |
HyukjinKwon
approved these changes
Sep 17, 2024
hmm, I don't think this is a “user-facing change” because: |
|
thanks, merged to master |
attilapiros
pushed a commit
to attilapiros/spark
that referenced
this pull request
Oct 4, 2024
### What changes were proposed in this pull request? Apply reservoir sampling in `SampledPlotBase` ### Why are the changes needed? Existing sampling approach has two drawbacks: 1, it needs two jobs to sample `max_rows` rows: - df.count() to compute `fraction = max_rows / count` - df.sample(fraction).to_pandas() to do the sampling 2, the df.sample is based on Bernoulli sampling which **cannot** guarantee the sampled size == expected `max_rows`, e.g. ``` In [1]: df = spark.range(10000) In [2]: [df.sample(0.01).count() for i in range(0, 10)] Out[2]: [96, 97, 95, 97, 105, 105, 105, 87, 95, 110] ``` The size of sampled data is floating near the target size 10000*0.01=100. This relative deviation cannot be ignored, when the input dataset is large and the sampling fraction is small. ### Does this PR introduce _any_ user-facing change? No ### How was this patch tested? CI and manually check ### Was this patch authored or co-authored using generative AI tooling? No Closes apache#48105 from zhengruifeng/ps_sampling. Authored-by: Ruifeng Zheng <ruifengz@apache.org> Signed-off-by: Ruifeng Zheng <ruifengz@apache.org>
himadripal
pushed a commit
to himadripal/spark
that referenced
this pull request
Oct 19, 2024
### What changes were proposed in this pull request? Apply reservoir sampling in `SampledPlotBase` ### Why are the changes needed? Existing sampling approach has two drawbacks: 1, it needs two jobs to sample `max_rows` rows: - df.count() to compute `fraction = max_rows / count` - df.sample(fraction).to_pandas() to do the sampling 2, the df.sample is based on Bernoulli sampling which **cannot** guarantee the sampled size == expected `max_rows`, e.g. ``` In [1]: df = spark.range(10000) In [2]: [df.sample(0.01).count() for i in range(0, 10)] Out[2]: [96, 97, 95, 97, 105, 105, 105, 87, 95, 110] ``` The size of sampled data is floating near the target size 10000*0.01=100. This relative deviation cannot be ignored, when the input dataset is large and the sampling fraction is small. ### Does this PR introduce _any_ user-facing change? No ### How was this patch tested? CI and manually check ### Was this patch authored or co-authored using generative AI tooling? No Closes apache#48105 from zhengruifeng/ps_sampling. Authored-by: Ruifeng Zheng <ruifengz@apache.org> Signed-off-by: Ruifeng Zheng <ruifengz@apache.org>
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
What changes were proposed in this pull request?
Apply reservoir sampling in
SampledPlotBaseWhy are the changes needed?
Existing sampling approach has two drawbacks:
1, it needs two jobs to sample
max_rowsrows:fraction = max_rows / count2, the df.sample is based on Bernoulli sampling which cannot guarantee the sampled size == expected
max_rows, e.g.The size of sampled data is floating near the target size 10000*0.01=100.
This relative deviation cannot be ignored, when the input dataset is large and the sampling fraction is small.
Does this PR introduce any user-facing change?
No
How was this patch tested?
CI and manually check
Was this patch authored or co-authored using generative AI tooling?
No