You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by "stankiewicz (via GitHub)" <gi...@apache.org> on 2025/01/28 14:40:12 UTC
[PR] Sample reporting as observer is sampling distribution [beam]
stankiewicz opened a new pull request, #33780:
URL: https://github.com/apache/beam/pull/33780
For Dataflow V2, StateBackedIterable is iterated by readers after gbk shuffle. Examples are ParDo after GBK or merging combiners after GBK.
[metrics.proto](https://github.com/apache/beam/blob/master/model/pipeline/src/main/proto/org/apache/beam/model/pipeline/v1/metrics.proto#L244) specifies that Sampling is used because calculating the byte count involves serializing the elements which is CPU intensive.
In case of StateBackedIterable sampling is not occurring which impacts performance of some of the pipelines that have expensive coders.
This change introduces sampling.
Fully fixes #33620 as previous fix was improvement.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Sample reporting as observer is sampling distribution [beam]
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
scwhittle commented on code in PR #33780:
URL: https://github.com/apache/beam/pull/33780#discussion_r1939007061
##########
sdks/java/harness/src/main/java/org/apache/beam/fn/harness/state/StateBackedIterable.java:
##########
@@ -138,15 +157,19 @@ public boolean hasNext() {
public T next() {
T value = wrappedIterator.next();
try {
- elementCoder.registerByteSizeObserver(value, observerProxy);
- if (observerProxy.getIsLazy()) {
- // The observer will only be notified of bytes as the result
- // is used. We defer advancing the observer until hasNext in an
- // attempt to capture those bytes.
- observerNeedsAdvance = true;
- } else {
- observerNeedsAdvance = false;
- observerProxy.advance();
+ if (sampleElement() || elementCoder.isRegisterByteSizeObserverCheap(value)) {
Review Comment:
if the byte observer is cheap, doesn't that mean we are scaling the elements below even if we are observing every time?
similarly it seems wrong if some values are cheap to observe and others are not.
Maybe it would be better as:
if (observerIsCheap(value)) {
update w/o scaling
} else if (sampleElement()) {
update w/ scaling
}
Since we're not calling sample() in case where it is cheap we wouldn't increase samplingToken except for elements that might be sampled.
this also seems like it might be a problem with OutputObjectAndByteCounter.java
##########
sdks/java/harness/src/test/java/org/apache/beam/fn/harness/state/StateBackedIterableTest.java:
##########
@@ -269,7 +269,8 @@ public void testByteObservingStateBackedIterable() throws Exception {
.sum();
observer.advance();
// 5 comes from size and hasNext (see IterableLikeCoder)
- assertEquals(iterateBytes + 5, observer.total);
+ // observer receives scaled
+ assertTrue(iterateBytes + 5 >= observer.total);
Review Comment:
is it less than 10 elements? shoudln't it be exact in that case?
If it's off by some double-math errors you could use some fuzzy equals (https://guava.dev/releases/23.0/api/docs/com/google/common/math/DoubleMath.html#fuzzyEquals-double-double-double- for example but perhaps some testing assert variant exists)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Sample reporting as observer is sampling distribution [beam]
Posted by "stankiewicz (via GitHub)" <gi...@apache.org>.
stankiewicz commented on PR #33780:
URL: https://github.com/apache/beam/pull/33780#issuecomment-2631233892
Run Java_Examples_Dataflow PreCommit
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Sample reporting as observer is sampling distribution [beam]
Posted by "stankiewicz (via GitHub)" <gi...@apache.org>.
stankiewicz commented on PR #33780:
URL: https://github.com/apache/beam/pull/33780#issuecomment-2631874375
Run Java PreCommit
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Sample reporting as observer is sampling distribution [beam]
Posted by "stankiewicz (via GitHub)" <gi...@apache.org>.
stankiewicz commented on PR #33780:
URL: https://github.com/apache/beam/pull/33780#issuecomment-2632054654
Run Java PreCommit
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Sample reporting as observer is sampling distribution [beam]
Posted by "stankiewicz (via GitHub)" <gi...@apache.org>.
stankiewicz commented on PR #33780:
URL: https://github.com/apache/beam/pull/33780#issuecomment-2630444778
Run Java_PVR_Flink_Docker PreCommit
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Sample reporting as observer is sampling distribution [beam]
Posted by "stankiewicz (via GitHub)" <gi...@apache.org>.
stankiewicz commented on PR #33780:
URL: https://github.com/apache/beam/pull/33780#issuecomment-2631873297
Run Java_Examples_Dataflow PreCommit
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Sample reporting as observer is sampling distribution [beam]
Posted by "stankiewicz (via GitHub)" <gi...@apache.org>.
stankiewicz commented on PR #33780:
URL: https://github.com/apache/beam/pull/33780#issuecomment-2631230078
Run Java_PVR_Flink_Docker PreCommit
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Sample reporting as observer is sampling distribution [beam]
Posted by "stankiewicz (via GitHub)" <gi...@apache.org>.
stankiewicz commented on PR #33780:
URL: https://github.com/apache/beam/pull/33780#issuecomment-2632115170
Run Java PreCommit
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Sample reporting as observer is sampling distribution [beam]
Posted by "stankiewicz (via GitHub)" <gi...@apache.org>.
stankiewicz commented on code in PR #33780:
URL: https://github.com/apache/beam/pull/33780#discussion_r1939075413
##########
sdks/java/harness/src/test/java/org/apache/beam/fn/harness/state/StateBackedIterableTest.java:
##########
@@ -269,7 +269,8 @@ public void testByteObservingStateBackedIterable() throws Exception {
.sum();
observer.advance();
// 5 comes from size and hasNext (see IterableLikeCoder)
- assertEquals(iterateBytes + 5, observer.total);
+ // observer receives scaled
+ assertTrue(iterateBytes + 5 >= observer.total);
Review Comment:
it's more than 10 elements, added assert with tolerance.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Sample reporting as observer is sampling distribution [beam]
Posted by "stankiewicz (via GitHub)" <gi...@apache.org>.
stankiewicz commented on PR #33780:
URL: https://github.com/apache/beam/pull/33780#issuecomment-2631229730
Run Java PreCommit
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Calculate byte size via sampling in StateBackedIterable if size is not cheap to calculate [beam]
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
scwhittle merged PR #33780:
URL: https://github.com/apache/beam/pull/33780
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Sample reporting as observer is sampling distribution [beam]
Posted by "stankiewicz (via GitHub)" <gi...@apache.org>.
stankiewicz commented on PR #33780:
URL: https://github.com/apache/beam/pull/33780#issuecomment-2631892660
Run Java_PVR_Flink_Docker PreCommit
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Sample reporting as observer is sampling distribution [beam]
Posted by "stankiewicz (via GitHub)" <gi...@apache.org>.
stankiewicz commented on PR #33780:
URL: https://github.com/apache/beam/pull/33780#issuecomment-2631936851
Run Java PreCommit
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Sample reporting as observer is sampling distribution [beam]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #33780:
URL: https://github.com/apache/beam/pull/33780#issuecomment-2620994118
Assigning reviewers. If you would like to opt out of this review, comment `assign to next reviewer`:
R: @damondouglas for label java.
Available commands:
- `stop reviewer notifications` - opt out of the automated review tooling
- `remind me after tests pass` - tag the comment author after tests pass
- `waiting on author` - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)
The PR bot will only process comments in the main thread (not review comments).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Sample reporting as observer is sampling distribution [beam]
Posted by "stankiewicz (via GitHub)" <gi...@apache.org>.
stankiewicz commented on PR #33780:
URL: https://github.com/apache/beam/pull/33780#issuecomment-2631873137
Run Java_PVR_Flink_Docker PreCommit
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Sample reporting as observer is sampling distribution [beam]
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
scwhittle commented on code in PR #33780:
URL: https://github.com/apache/beam/pull/33780#discussion_r1939108661
##########
sdks/java/harness/src/main/java/org/apache/beam/fn/harness/state/StateBackedIterable.java:
##########
@@ -105,6 +106,12 @@ private static class WrappedObservingIterator<T> extends ElementByteSizeObservab
private boolean observerNeedsAdvance = false;
private boolean exceptionLogged = false;
+ private final Random randomGenerator = new Random();
Review Comment:
should we use ThreadLocalRandom?
for dataflow streaming we have lots of bundles, and thus might have a lot of these if used for input.
see https://github.com/apache/beam/pull/33737 for OutputObjectAndByteCounter change
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Sample reporting as observer is sampling distribution [beam]
Posted by "stankiewicz (via GitHub)" <gi...@apache.org>.
stankiewicz commented on PR #33780:
URL: https://github.com/apache/beam/pull/33780#issuecomment-2620091188
Run Java PreCommit
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Sample reporting as observer is sampling distribution [beam]
Posted by "stankiewicz (via GitHub)" <gi...@apache.org>.
stankiewicz commented on PR #33780:
URL: https://github.com/apache/beam/pull/33780#issuecomment-2632103624
flaky tests: https://develocity.apache.org/scans/tests?search.timeZoneId=Europe%2FWarsaw&tests.container=org.apache.beam.runners.dataflow.worker.windmill.client.grpc.GrpcDirectGetWorkStreamTest&tests.test=testConsumedWorkItems_itemsSplitAcrossResponses&tests.unstableOnly=false
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Sample reporting as observer is sampling distribution [beam]
Posted by "stankiewicz (via GitHub)" <gi...@apache.org>.
stankiewicz commented on code in PR #33780:
URL: https://github.com/apache/beam/pull/33780#discussion_r1939084138
##########
sdks/java/harness/src/main/java/org/apache/beam/fn/harness/state/StateBackedIterable.java:
##########
@@ -138,15 +157,19 @@ public boolean hasNext() {
public T next() {
T value = wrappedIterator.next();
try {
- elementCoder.registerByteSizeObserver(value, observerProxy);
- if (observerProxy.getIsLazy()) {
- // The observer will only be notified of bytes as the result
- // is used. We defer advancing the observer until hasNext in an
- // attempt to capture those bytes.
- observerNeedsAdvance = true;
- } else {
- observerNeedsAdvance = false;
- observerProxy.advance();
+ if (sampleElement() || elementCoder.isRegisterByteSizeObserverCheap(value)) {
Review Comment:
done. I will fix OutputObjectAndByteCounter in next pr.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org