scwhittle commented on code in PR #38783:
URL: https://github.com/apache/beam/pull/38783#discussion_r4216734988
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/StorageApiLoads.java:
##########
@@ -379,12 +381,19 @@ public WriteResult expandUntriggered(
PCollection<KV<DestinationT, StorageApiWritePayload>>
successfulConvertedRows =
convertMessagesResult.get(successfulConvertedRowsTag);
- if (numShards > 0) {
+ boolean streaming =
input.getPipeline().getOptions().as(StreamingOptions.class).isStreaming();
+ if (numShards > 0 && streaming) {
successfulConvertedRows =
successfulConvertedRows.apply(
"ResdistibuteNumShards",
Redistribute.<KV<DestinationT,
StorageApiWritePayload>>arbitrarily()
.withNumBuckets(numShards));
+ } else if (numShards > 0 && !streaming) {
+ successfulConvertedRows =
+ successfulConvertedRows
+ .apply("AddKeyWithSideInputs", ParDo.of(new
AddShardKeyFn<>(numShards)))
Review Comment:
It seems like where the current behavior might be ok and the new behavior
bad is if there are pipelines with many sparse destinations and a few heavy
destinations. In that case the # of keys will greatly increase.
Maybe we could consider:
- having the # of shards be part of the dynamic destination config? Then
users can choose to shard destinations that are large but not small ones
- adding a new option on if shards should be per-destination or overall
(perhaps defaulting to true). Then if there are issues with the new logic for
pipelines with sparse destinatoins and heavy hitters requiring sharding it can
easily be disabled.
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/StorageApiLoads.java:
##########
@@ -379,12 +381,19 @@ public WriteResult expandUntriggered(
PCollection<KV<DestinationT, StorageApiWritePayload>>
successfulConvertedRows =
convertMessagesResult.get(successfulConvertedRowsTag);
- if (numShards > 0) {
+ boolean streaming =
input.getPipeline().getOptions().as(StreamingOptions.class).isStreaming();
+ if (numShards > 0 && streaming) {
successfulConvertedRows =
successfulConvertedRows.apply(
"ResdistibuteNumShards",
Redistribute.<KV<DestinationT,
StorageApiWritePayload>>arbitrarily()
.withNumBuckets(numShards));
+ } else if (numShards > 0 && !streaming) {
+ successfulConvertedRows =
+ successfulConvertedRows
+ .apply("AddKeyWithSideInputs", ParDo.of(new
AddShardKeyFn<>(numShards)))
Review Comment:
why does this have WithSideInputs in the transform name?
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
