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
...am/batch-write-flow-control-example/src/main/java/bigtable/BatchWriteFlowControlExample.java
Show resolved
Hide resolved
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.
There was a problem hiding this comment.
This code sample shows how to enable the flow control feature. It has 2 parts because we have two different connectors and they're configured differently. I was unsure whether it'll worth duplicating the rest of the code to keep the sample simple. Please advice and I can split.
We're going to write the doc that will use this code sample.
Yes I'll add 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.
Does it intend for run asynchronously? Please, append waitUntilFinish() call to the result of the run().
There was a problem hiding this comment.
So far our example jobs all run async, is it a better practice to run sync? I'm happy to learn the reasoning of the practice.
Includes using BigtableIO and CloudBigtableIO
kongweihan
force-pushed
the
flow-control-example
branch
from
d563567
to
b29139e
Compare
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