slilichenko commented on code in PR #25723:
URL: https://github.com/apache/beam/pull/25723#discussion_r1126846928
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/StorageApiWriteUnshardedRecords.java:
##########
@@ -558,7 +574,23 @@ long flush(
appendFailures.inc();
return RetryType.RETRY_ALL_OPERATIONS;
},
- c -> recordsAppended.inc(c.protoRows.getSerializedRowsCount()),
+ c -> {
+ recordsAppended.inc(c.protoRows.getSerializedRowsCount());
+ if (successfulRowsReceiver != null) {
+ for (ByteString rowBytes :
c.protoRows.getSerializedRowsList()) {
+ try {
+ TableRow row =
Review Comment:
This will restore the whole row and could be costly to just carry the whole
thing around for the sole purpose of Wait.on(). Is it easy to return a row of a
list of fields that can potentially be needed (something that constitutes a
primary key of the record)?
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/StorageApiWriteUnshardedRecords.java:
##########
@@ -558,7 +574,23 @@ long flush(
appendFailures.inc();
return RetryType.RETRY_ALL_OPERATIONS;
},
- c -> recordsAppended.inc(c.protoRows.getSerializedRowsCount()),
+ c -> {
+ recordsAppended.inc(c.protoRows.getSerializedRowsCount());
+ if (successfulRowsReceiver != null) {
+ for (ByteString rowBytes :
c.protoRows.getSerializedRowsList()) {
+ try {
+ TableRow row =
+ TableRowToStorageApiProto.tableRowFromMessage(
+ DynamicMessage.parseFrom(
+
Preconditions.checkStateNotNull(appendClientInfo).getDescriptor(),
+ rowBytes));
+ successfulRowsReceiver.output(row);
Review Comment:
This assumes that there are no re-windowing of the data into the Global
Window in the previous steps. Is it correct?
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/StorageApiWriteUnshardedRecords.java:
##########
@@ -758,7 +800,24 @@ public void outputWithTimestamp(
BigQueryStorageApiInsertError output, org.joda.time.Instant
timestamp) {
context.output(failedRowsTag, output, timestamp,
GlobalWindow.INSTANCE);
}
- });
+ };
+ @Nullable OutputReceiver<TableRow> successfulRowsReceiver = null;
+ if (successfulRowsTag != null) {
+ successfulRowsReceiver =
+ new OutputReceiver<TableRow>() {
Review Comment:
It looks like the original windows are not preserved here. The assumption is
that most of the time the customer will use FixedWindows in streaming pipelines.
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/StorageApiWriteUnshardedRecords.java:
##########
@@ -558,7 +574,23 @@ long flush(
appendFailures.inc();
return RetryType.RETRY_ALL_OPERATIONS;
},
- c -> recordsAppended.inc(c.protoRows.getSerializedRowsCount()),
+ c -> {
+ recordsAppended.inc(c.protoRows.getSerializedRowsCount());
+ if (successfulRowsReceiver != null) {
+ for (ByteString rowBytes :
c.protoRows.getSerializedRowsList()) {
+ try {
+ TableRow row =
+ TableRowToStorageApiProto.tableRowFromMessage(
+ DynamicMessage.parseFrom(
+
Preconditions.checkStateNotNull(appendClientInfo).getDescriptor(),
+ rowBytes));
+ successfulRowsReceiver.output(row);
+ } catch (InvalidProtocolBufferException e) {
+ LOG.warn("Failure parsing TableRow: " + e);
Review Comment:
Stack trace is lost; would be good to have it. Would be useful to add a
custom metric to track these errors, otherwise there is no way to know if these
error occur. This failure can also affect the pipeline correctness. If all
records fail to convert - none of the windows will show up. A couple of options
to make it more robust - carry customer defined "record key" through the whole
processing and output that instead of the full collection or output a TableRow
with a "failed-to-convert" column/value and the same window as the original.
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/StorageApiWritesShardedRecords.java:
##########
@@ -647,6 +662,12 @@ public void process(
context.offset +
context.protoRows.getSerializedRowsCount() - 1,
false)));
flushesScheduled.inc(context.protoRows.getSerializedRowsCount());
+
+ if (successfulRowsTag != null) {
+ for (ByteString protoBytes :
context.protoRows.getSerializedRowsList()) {
+
o.get(successfulRowsTag).output(appendClientInfo.get().toTableRow(protoBytes));
Review Comment:
can toTableRow() throw an exception here? A different conversion approach is
used in another place (DynamicMessage.parseFrom(...)).
--
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]
