Add a batch write flow control example for Bigtable #9314
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
product-auto-label
bot
added
samples
kongweihan
force-pushed
the
flow-control-example
branch
5 times, most recently
from
411e662
to
6e4f7db
Compare
| @@ -0,0 +1,127 @@ | |||
| <?xml version="1.0" encoding="UTF-8"?> | |||
| <!-- | |||
| Copyright 2021 Google LLC | |||
|
|
||
| PCollection<Long> numbers = p.apply(generateLabel, GenerateSequence.from(0).to(numRows)); | ||
|
|
||
| if (options.getUseCloudBigtableIo()) { |
There was a problem hiding this comment.
do we need the code showing both ways for this? I know we recommend customers use one of these over the other but can never remember which one. Ideally we just show the recommended way. Or if the goal is to just show how to activate the flow control on the client then we could keep both but maybe the data being written can be simplified to not have as much information for customers to take in.
For example, we could just write one row using each client and that can reduce the need to generate sequences and all the additional helper functions.
kongweihan
force-pushed
the
flow-control-example
branch
from
6e4f7db
to
d563567
Compare
minherz
requested changes
May 23, 2024
There was a problem hiding this comment.
Hello,
Please address the following questions:
- We ask to have a single code sample per file. This code looks like showing two code samples. What is this code sample demonstrates?
- This code sample does not have regional tags. What documentation use it?
- We do not host code samples without tests. What is a reason for lack of tests?
Comment on lines
+86
to
+126
| Pipeline p = Pipeline.create(options); | ||
|
|
||
| PCollection<Long> numbers = p.apply(generateLabel, GenerateSequence.from(0).to(numRows)); | ||
|
|
||
| if (options.getUseCloudBigtableIo()) { | ||
| System.out.println("Using CloudBigtableIO"); | ||
| PCollection<org.apache.hadoop.hbase.client.Mutation> mutations = numbers.apply(mutationLabel, | ||
| ParDo.of(new CreateHbaseMutationFn(options.getBigtableColsPerRow(), | ||
| options.getBigtableBytesPerCol()))); | ||
|
|
||
| mutations.apply( | ||
| String.format("Write data to table %s via CloudBigtableIO", options.getBigtableTableId()), | ||
| CloudBigtableIO.writeToTable(new CloudBigtableTableConfiguration.Builder() | ||
| .withProjectId(options.getProject()) | ||
| .withInstanceId(options.getBigtableInstanceId()) | ||
| .withTableId(options.getBigtableTableId()) | ||
| .withConfiguration(BigtableOptionsFactory.BIGTABLE_ENABLE_BULK_MUTATION_FLOW_CONTROL, | ||
| "true") | ||
| .withConfiguration(BigtableOptionsFactory.BIGTABLE_BULK_MAX_REQUEST_SIZE_BYTES, | ||
| "1048576") | ||
| .build())); | ||
| } else { | ||
| System.out.println("Using BigtableIO"); | ||
| PCollection<KV<ByteString, Iterable<Mutation>>> | ||
| mutations = numbers.apply(mutationLabel, | ||
| ParDo.of(new CreateMutationFn(options.getBigtableColsPerRow(), | ||
| options.getBigtableBytesPerCol()))); | ||
|
|
||
| BigtableIO.Write write = BigtableIO.write() | ||
| .withProjectId(options.getProject()) | ||
| .withInstanceId(options.getBigtableInstanceId()) | ||
| .withTableId(options.getBigtableTableId()) | ||
| .withFlowControl(true); // This enables batch write flow control | ||
|
|
||
| mutations.apply( | ||
| String.format("Write data to table %s via BigtableIO", options.getBigtableTableId()), | ||
| write | ||
| ); | ||
| } | ||
|
|
||
| p.run(); |
There was a problem hiding this comment.
this block of code is hard to read. Since it is a code sample it should be easy to understand. Please, reformat it so it will look like steps each of the steps calling apply method of the pipeline. See the dataflow-bigquery-read-tablerows sample as a reference.
Comment on lines
+86
to
+126
| Pipeline p = Pipeline.create(options); | ||
|
|
||
| PCollection<Long> numbers = p.apply(generateLabel, GenerateSequence.from(0).to(numRows)); | ||
|
|
||
| if (options.getUseCloudBigtableIo()) { | ||
| System.out.println("Using CloudBigtableIO"); | ||
| PCollection<org.apache.hadoop.hbase.client.Mutation> mutations = numbers.apply(mutationLabel, | ||
| ParDo.of(new CreateHbaseMutationFn(options.getBigtableColsPerRow(), | ||
| options.getBigtableBytesPerCol()))); | ||
|
|
||
| mutations.apply( | ||
| String.format("Write data to table %s via CloudBigtableIO", options.getBigtableTableId()), | ||
| CloudBigtableIO.writeToTable(new CloudBigtableTableConfiguration.Builder() | ||
| .withProjectId(options.getProject()) | ||
| .withInstanceId(options.getBigtableInstanceId()) | ||
| .withTableId(options.getBigtableTableId()) | ||
| .withConfiguration(BigtableOptionsFactory.BIGTABLE_ENABLE_BULK_MUTATION_FLOW_CONTROL, | ||
| "true") | ||
| .withConfiguration(BigtableOptionsFactory.BIGTABLE_BULK_MAX_REQUEST_SIZE_BYTES, | ||
| "1048576") | ||
| .build())); | ||
| } else { | ||
| System.out.println("Using BigtableIO"); | ||
| PCollection<KV<ByteString, Iterable<Mutation>>> | ||
| mutations = numbers.apply(mutationLabel, | ||
| ParDo.of(new CreateMutationFn(options.getBigtableColsPerRow(), | ||
| options.getBigtableBytesPerCol()))); | ||
|
|
||
| BigtableIO.Write write = BigtableIO.write() | ||
| .withProjectId(options.getProject()) | ||
| .withInstanceId(options.getBigtableInstanceId()) | ||
| .withTableId(options.getBigtableTableId()) | ||
| .withFlowControl(true); // This enables batch write flow control | ||
|
|
||
| mutations.apply( | ||
| String.format("Write data to table %s via BigtableIO", options.getBigtableTableId()), | ||
| write | ||
| ); | ||
| } | ||
|
|
||
| p.run(); |
There was a problem hiding this comment.
Does it intend for run asynchronously? Please, append waitUntilFinish() call to the result of the run().
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.
Description
Add a batch write flow control example for Bigtable
Checklist
pom.xmlparent set to latestshared-configurationmvn clean verifyrequiredmvn -P lint checkstyle:checkrequiredmvn -P lint clean compile pmd:cpd-check spotbugs:checkadvisory only