Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

PCollection data sampling for Java SDK harness #25064 #25354

Merged
merged 43 commits into from
Feb 22, 2023
Merged
Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
43 commits
Select commit Hold shift + click to select a range
cd630e4
Data Sampling Java Impl
Jan 26, 2023
e530d14
comments
Jan 26, 2023
a09f5d6
add PBD id to context
rohdesamuel Jan 31, 2023
1cb08f9
merge
Jan 31, 2023
566c7b8
Add more tests and spotless
Jan 31, 2023
7395a2e
Finish Java data sampling impl with tests, adding comments
Feb 1, 2023
4c1253f
more comments, remove Payload
Feb 1, 2023
3d88254
more comments
Feb 1, 2023
5bbef91
spotless
Feb 1, 2023
6993b83
Encode in the nested context
rohdesamuel Feb 3, 2023
769902f
Update sdks/java/harness/src/main/java/org/apache/beam/fn/harness/con…
rohdesamuel Feb 7, 2023
5c06d4e
Apply suggestions from code review
rohdesamuel Feb 7, 2023
99087e8
address pr comments
Feb 9, 2023
5609e4a
give default pbd id to test context
rohdesamuel Feb 9, 2023
694c929
address spotlesscheck
rohdesamuel Feb 9, 2023
519aece
spotless apply
rohdesamuel Feb 9, 2023
8efbb15
style guide spotless apply
rohdesamuel Feb 9, 2023
cd3732d
add serviceloader
rohdesamuel Feb 9, 2023
415e3f0
change datasampling to modify the consumers and not graph for sampling
rohdesamuel Feb 10, 2023
553fc5e
remove redundant SamplerState obj
rohdesamuel Feb 10, 2023
e260eb4
spotless
rohdesamuel Feb 10, 2023
4f29308
replace mutex with atomics in output sampler to reduce contention
rohdesamuel Feb 10, 2023
8cbc6c8
spotless and fix OutputSamplerTest
rohdesamuel Feb 10, 2023
f67234f
Update sdks/java/harness/src/main/java/org/apache/beam/fn/harness/dat…
rohdesamuel Feb 13, 2023
6815330
Update sdks/java/harness/src/main/java/org/apache/beam/fn/harness/dat…
rohdesamuel Feb 13, 2023
4848725
Update sdks/java/harness/src/main/java/org/apache/beam/fn/harness/deb…
rohdesamuel Feb 13, 2023
6c16576
always init outputsampler
rohdesamuel Feb 13, 2023
822587d
add final to DataSampler in FnHarness
rohdesamuel Feb 13, 2023
fe9ab2a
spotless apply
rohdesamuel Feb 13, 2023
07d37ea
update from proto names
rohdesamuel Feb 13, 2023
5e9c4b0
spotless bugs
Feb 14, 2023
f5f97fb
Apply suggestions from code review
rohdesamuel Feb 14, 2023
c3db7c0
address pr comments
Feb 14, 2023
fce6d69
spotlessapply and add byte[] test
Feb 15, 2023
69d8bb4
validate datasampler args
Feb 15, 2023
be2ebe4
add concurrency tests
Feb 15, 2023
93af06e
Merge branch 'master' into data-sampling-java
lukecwik Feb 21, 2023
cbdbbc3
Update sdks/java/harness/src/main/java/org/apache/beam/fn/harness/deb…
rohdesamuel Feb 21, 2023
8cc8ee0
Update sdks/java/harness/src/test/java/org/apache/beam/fn/harness/deb…
rohdesamuel Feb 21, 2023
10aa7de
Update sdks/java/harness/src/test/java/org/apache/beam/fn/harness/deb…
rohdesamuel Feb 21, 2023
04c6c88
Update sdks/java/harness/src/test/java/org/apache/beam/fn/harness/deb…
rohdesamuel Feb 21, 2023
862439b
improve contention tests
Feb 22, 2023
490be4a
spotless
Feb 22, 2023
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Prev Previous commit
Next Next commit
Apply suggestions from code review
Co-authored-by: Lukasz Cwik <lcwik@google.com>
  • Loading branch information
rohdesamuel and lukecwik committed Feb 13, 2023
commit 5c06d4ebda7aa236a1ff4eae56cd4bfd36a1712e
Original file line number Diff line number Diff line change
Expand Up @@ -51,10 +51,10 @@ public DataSampler(int maxSamples, int sampleEveryN) {
}

// Maximum number of elements in buffer.
private int maxSamples = 10;
private final int maxSamples = 10;

// Sampling rate.
private int sampleEveryN = 1000;
private final int sampleEveryN = 1000;

// The fully-qualified type is: Map[ProcessBundleDescriptorId, [PCollectionId, OutputSampler]].
// The DataSampler object lives on the same level of the FnHarness. This means that many threads
Expand Down Expand Up @@ -142,7 +142,7 @@ public Map<String, List<byte[]>> samplesFor(Set<String> descriptors, Set<String>
return;
}

samples.putIfAbsent(pcollectionId, new ArrayList<>());
samples.putIfAbsent(pcollectionId, Collections.EMPTY_LIST);
samples.get(pcollectionId).addAll(outputSampler.samples());
});
});
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -33,7 +33,7 @@
*/
public class OutputSampler<T> {
private final Coder<T> coder;
private final List<T> buffer = new ArrayList<>();
private final List<T> buffer = new ArrayList<>(maxElements);
private static final Logger LOG = LoggerFactory.getLogger(OutputSampler.class);

// Maximum number of elements in buffer.
Expand Down