Support Avro GenericRecord as a valid format for StorageWrite API on BigQueryIO #24274

prodriguezdefino · 2022-11-18T23:35:05Z

This change enables the use of AVRO GenericRecord as part of the BigQueryIO storage write implementation (consistent and at-least-once).

When using AVRO GenericRecord as the transport object in their pipelines, users need to configure the use of Beam schemas in the BigQueryIO to transform their data into Beam Rows first, or provide a format function to transform into TableRow, to then have their data formatted as DynamicMessage protos for BQ StorageWrite to work. This PR enables the users to provide a schema (BQ TableSchema, dyn destination or schema from view) and an AvroRowFactoryWriter to have their data transformed to GenericRecord and then StorageWrite compatible protos.

When using GenericRecord as the input type for the PCollection, providing the identity function should be sufficient and inexpensive for their data ingestion.

Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:

Choose reviewer(s) and mention them in a comment (R: @username).
Mention the appropriate issue in your description (for example: addresses #123), if applicable. This will automatically add a link to the pull request in the issue. If you would like the issue to automatically close on merging the pull request, comment fixes #<ISSUE NUMBER> instead.
Update CHANGES.md with noteworthy changes.
If this contribution is large, please file an Apache Individual Contributor License Agreement.

See the Contributor Guide for more tips on how to make review process smoother.

To check the build health, please visit https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md

GitHub Actions Tests Status (on master branch)

See CI.md for more information about GitHub Actions CI.

github-actions · 2022-11-20T03:47:55Z

Checks are failing. Will not request review until checks are succeeding. If you'd like to override that behavior, comment assign set of reviewers

github-actions · 2022-11-21T19:04:57Z

Assigning reviewers. If you would like to opt out of this review, comment assign to next reviewer:

R: @kileys for label java.
R: @Abacn for label io.

Available commands:

stop reviewer notifications - opt out of the automated review tooling
remind me after tests pass - tag the comment author after tests pass
waiting on author - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)

The PR bot will only process comments in the main thread (not review comments).

prodriguezdefino · 2022-11-21T20:50:10Z

Run Java PreCommit

prodriguezdefino · 2022-11-23T09:48:47Z

Run Java PreCommit

prodriguezdefino · 2022-11-23T09:54:46Z

Fixes #24329

github-actions · 2022-11-30T12:14:03Z

Reminder, please take a look at this pr: @kileys @Abacn

Abacn · 2022-11-30T16:16:52Z

Thanks for the contribution! Taking a look.

sdks/java/core/src/main/java/org/apache/beam/sdk/schemas/utils/AvroUtils.java

Abacn

Thanks for the work! Looks solid to me. Haven't get into two proto related new classes yet. Left a few comments found.

Abacn · 2022-11-30T21:20:11Z

...o/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryUtils.java

          .put(SqlTypes.DATETIME.getIdentifier(), StandardSQLTypeName.DATETIME)
          .put("SqlTimeWithLocalTzType", StandardSQLTypeName.TIME)
          .put("Enum", StandardSQLTypeName.STRING)
+          .put(MicrosInstant.IDENTIFIER, StandardSQLTypeName.TIMESTAMP)


Is there a test verifying that this logical type is supported? MicrosInstant is used for java.time.Instant (saying, testing write java.time.Instant to BigQuery)

MicrosInstant is also used for the translation of the Avro logical type named TimestampMicros (which is being forced in the test BigQueryWriteIOTest.testWriteAvro().
For the time related logical types (timestampmicros and millis, alongside with Date), Avro types store internally a numeric value that represents time since epoch (micros, millis or days depending on the type).
I'm not sure if I can force an Avro object to carry a java.time.Instant (maybe not) and this code path executes only when the underlying Avro GenericRecord has a time related logical type (which is being tested) and its schema gets translated to a Beam Row.Schema yielding the MicrosInstant (AvroUtils.toFieldType method).

fyi Support of Micros_instant in AvroUtils is about to rolled back due to a breaking change (#24489). By now MicrosInstant logical type (and java.time.Instant) support is incomplete throughout the code base. Sorry for inconvenience.

oh, so the only supported time related logical types will be millis and date for now, is that right?
I can remove the timestamp in micros for the time being if that works for this PR. sounds good?

yeah sounds good to me

ok, in fact I rebased from master and removed the MicrosInstant mapping and the tests are still running fine. So I will keep the micros support since now those get mapped to Instant and work fine.

reuvenlax · 2022-11-30T22:16:23Z

...a/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIO.java

          storageApiDynamicDestinations =
              new StorageApiDynamicDestinationsBeamRow<>(
                  dynamicDestinations, elementSchema, elementToRowFunction);
+        } else if (getAvroRowWriterFactory() != null) {


Please add a writeGenericRecords() method so users don't have to deal with the avro format function.

Also - AFAICT from inspection, the AvroWriterFactory is a bit broken if side inputs are used for schemas. Do we have a test for this?

Please add a writeGenericRecords() method so users don't have to deal with the avro format function.

Done.

AFAICT from inspection, the AvroWriterFactory is a bit broken if side inputs are used for schemas. Do we have a test for this?

You mean when the TableSchema is provided as a PCollectionView? If so I can add a test for that.

Yeah - because setSideInputAccessorFromProcessContext needs to be called on the dynamic destinations in every DoFn that might access this, and I'm not sure that's correct for AvroWriterFactory

Added a test to check on this for StorageWrites API

Also and JIC, besides the added unit test, checked this PR's code with a simple DynamicDestination definition and seems to work fine for job id: 2023-01-31_12_10_08-12389116961981372457

reuvenlax · 2022-11-30T22:22:06Z

Please resolve against current codebase

...ain/java/org/apache/beam/sdk/io/gcp/bigquery/StorageApiDynamicDestinationsGenericRecord.java

prodriguezdefino · 2022-12-01T17:22:26Z

Run Spotless PreCommit

prodriguezdefino · 2022-12-01T17:42:45Z

Run Java_GCP_IO_Direct PreCommit

prodriguezdefino · 2022-12-01T18:04:36Z

Run Java PreCommit

reuvenlax · 2022-12-02T21:19:54Z

...rm/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/AvroGenericRecordToStorageApiProto.java

+    Schema elementType = null;
+    switch (schema.getType()) {
+      case RECORD:
+        if (schema == null) {


If this were null, the above switch would already have thrown. Did you mean to check something else?

Good catch, moved to before the switch statement.

reuvenlax · 2022-12-02T21:20:23Z

...rm/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/AvroGenericRecordToStorageApiProto.java

+        if (schema == null) {
+          throw new RuntimeException("Unexpected null schema!");
+        }
+        Preconditions.checkState(!schema.getFields().isEmpty());


Are empty records disallowed in avro?

They are allowed, the check was also in the BeamRowToStorageApiProto similar methods (Beam Row.Schema to Descriptors) and wanted to put the same safeguards.

Not sure how useful would be to have an empty record in this context, but let me know if you want me to remove the restriction.

reuvenlax · 2022-12-02T21:26:35Z

...rm/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/AvroGenericRecordToStorageApiProto.java

+      case RECORD:
+        return messageFromGenericRecord(fieldDescriptor.getMessageType(), (GenericRecord) value);
+      case ARRAY:
+        List<Object> list = (List<Object>) value;


Are we sure it's always a List (and not just an Iterable)?

Changed it to Iterable.

...a/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIO.java

reuvenlax · 2022-12-02T21:30:24Z

...a/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIO.java

          storageApiDynamicDestinations =
              new StorageApiDynamicDestinationsBeamRow<>(
                  dynamicDestinations, elementSchema, elementToRowFunction);
+        } else if (getAvroRowWriterFactory() != null) {


Yeah - because setSideInputAccessorFromProcessContext needs to be called on the dynamic destinations in every DoFn that might access this, and I'm not sure that's correct for AvroWriterFactory

github-actions · 2023-01-06T12:13:39Z

Assigning new set of reviewers because Pr has gone too long without review. If you would like to opt out of this review, comment assign to next reviewer:

R: @robertwb for label java.
R: @pabloem for label io.

Available commands:

stop reviewer notifications - opt out of the automated review tooling
remind me after tests pass - tag the comment author after tests pass
waiting on author - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)

github-actions · 2023-01-13T12:14:20Z

Reminder, please take a look at this pr: @robertwb @pabloem

Abacn

Sorry for the delay. Will take another look to see if previous comments are resolved. Or @reuvenlax do you still have a chance to review, thanks!

sdks/java/core/src/main/java/org/apache/beam/sdk/schemas/utils/AvroUtils.java

github-actions · 2023-01-18T12:14:00Z

Assigning new set of reviewers because Pr has gone too long without review. If you would like to opt out of this review, comment assign to next reviewer:

R: @kileys for label java.
R: @johnjcasey for label io.

Available commands:

stop reviewer notifications - opt out of the automated review tooling
remind me after tests pass - tag the comment author after tests pass
waiting on author - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)

github-actions · 2023-01-26T12:14:22Z

Reminder, please take a look at this pr: @kileys @johnjcasey

github-actions · 2023-01-31T12:13:43Z

Assigning new set of reviewers because Pr has gone too long without review. If you would like to opt out of this review, comment assign to next reviewer:

R: @kennknowles for label java.
R: @pabloem for label io.

Available commands:

stop reviewer notifications - opt out of the automated review tooling
remind me after tests pass - tag the comment author after tests pass
waiting on author - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)

prodriguezdefino · 2023-02-01T01:04:49Z

Run Java_Kafka_IO_Direct PreCommit

prodriguezdefino · 2023-02-01T02:31:21Z

Tested this change with 2 very similar pipelines which:

read from PS
transform format into AVRO
and then one of them directly go into BQ and the other one transform to Row using beam schema (current possible path with code in master, the other one would be to write TableRow),
both pipelines process ~250MB/s
and as expected the difference in resource utilization is significant.

Using beam rows as BigQueryIO input format:

Using write GenericRecords as BigQueryIO input format:

The difference in vCPU utilization at similar runtime is 176 vCPU/hr vs 117 vCPU/hr, more than 40% improvement when using GenericRecords.

prodriguezdefino · 2023-02-01T02:31:42Z

Run Java_Kafka_IO_Direct PreCommit

github-actions · 2023-02-12T12:13:30Z

Reminder, please take a look at this pr: @kennknowles @pabloem

Abacn · 2023-02-14T17:05:59Z

Hi @prodriguezdefino thanks for the patience. This looks good to me. Would you mind addressing #24274 (comment) so that when #24992 also gets in it won't break? Thanks!

…xtensions module

Abacn

Thanks! Would you mind adding a piece of announcement in https://github.com/apache/beam/blob/master/CHANGES.md?plain=1#L62

will merge when test passed

prodriguezdefino · 2023-02-16T23:48:07Z

Run Java PreCommit

prodriguezdefino · 2023-02-16T23:48:28Z

Run Java_Kafka_IO_Direct PreCommit

Abacn · 2023-02-17T01:25:22Z

Test passed. Some GitHub action job hanging due to self-host runner issues. merging for now.

initial impl

685fc06

github-actions bot added gcp io java labels Nov 18, 2022

prodriguezdefino changed the title ~~initial impl~~ Support Avro GenericRecord as a valid format for StorageWrite API on BigQueryIO Nov 19, 2022

prodriguezdefino marked this pull request as ready for review November 20, 2022 03:29

force unit tests

4be8f9d

github-actions bot added the Next Action: Reviewers label Nov 21, 2022

github-actions bot added the slow-review label Nov 30, 2022

github-actions bot removed the slow-review label Nov 30, 2022

Abacn reviewed Nov 30, 2022

View reviewed changes

sdks/java/core/src/main/java/org/apache/beam/sdk/schemas/utils/AvroUtils.java Outdated Show resolved Hide resolved

Abacn reviewed Nov 30, 2022

View reviewed changes

reuvenlax reviewed Nov 30, 2022

View reviewed changes

Merge branch 'master' into avro_support_bq_storagewrite

7772e82

reuvenlax reviewed Dec 1, 2022

View reviewed changes

...ain/java/org/apache/beam/sdk/io/gcp/bigquery/StorageApiDynamicDestinationsGenericRecord.java Outdated Show resolved Hide resolved

...ain/java/org/apache/beam/sdk/io/gcp/bigquery/StorageApiDynamicDestinationsGenericRecord.java Outdated Show resolved Hide resolved

prodriguezdefino added 3 commits November 30, 2022 19:09

addressing comments

4dcf9a4

spotless and checkstyle checks fixes

ea52993

spotless and checkstyle checks fixes

962cf51

reuvenlax reviewed Dec 2, 2022

View reviewed changes

github-actions bot removed the slow-review label Jan 6, 2023

github-actions bot added the slow-review label Jan 13, 2023

Abacn reviewed Jan 13, 2023

View reviewed changes

sdks/java/core/src/main/java/org/apache/beam/sdk/schemas/utils/AvroUtils.java Show resolved Hide resolved

github-actions bot removed the slow-review label Jan 18, 2023

Merge branch 'apache:master' into avro_support_bq_storagewrite

60db300

github-actions bot added the slow-review label Jan 26, 2023

github-actions bot removed the slow-review label Jan 31, 2023

prodriguezdefino added a commit to prodriguezdefino/apache-beam-streaming-tests that referenced this pull request Jan 31, 2023

testing apache/beam#24274 and apache/beam#22179

9196d4a

prodriguezdefino mentioned this pull request Jan 31, 2023

Changes after Avro related changes on BigQueryIO prodriguezdefino/apache-beam-streaming-tests#1

Merged

prodriguezdefino added 2 commits January 31, 2023 12:23

Merge branch 'master' into avro_support_bq_storagewrite

5f571e5

merge from master, fix tests and validations

33c7974

github-actions bot added the slow-review label Feb 12, 2023

Abacn self-requested a review February 13, 2023 22:04

Abacn mentioned this pull request Feb 14, 2023

[Avro] Use "extensions/avro" instead of avro from"core" in Java SDK modules #24992

Merged

3 tasks

copying the changes in AvroUtils located in core module to the avro e…

fd50138

…xtensions module

Abacn approved these changes Feb 16, 2023

View reviewed changes

Abacn merged commit 40838f7 into apache:master Feb 17, 2023

Support Avro GenericRecord as a valid format for StorageWrite API on BigQueryIO #24274

Support Avro GenericRecord as a valid format for StorageWrite API on BigQueryIO #24274

Uh oh!

Conversation

prodriguezdefino commented Nov 18, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

GitHub Actions Tests Status (on master branch)

Uh oh!

github-actions bot commented Nov 20, 2022

Uh oh!

github-actions bot commented Nov 21, 2022

Uh oh!

prodriguezdefino commented Nov 21, 2022

Uh oh!

prodriguezdefino commented Nov 23, 2022

Uh oh!

prodriguezdefino commented Nov 23, 2022

Uh oh!

github-actions bot commented Nov 30, 2022

Uh oh!

Abacn commented Nov 30, 2022

Uh oh!

Uh oh!

Abacn left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

prodriguezdefino Dec 1, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Abacn Dec 2, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

prodriguezdefino Dec 1, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

reuvenlax commented Nov 30, 2022

Uh oh!

Uh oh!

Uh oh!

prodriguezdefino commented Dec 1, 2022

Uh oh!

prodriguezdefino commented Dec 1, 2022

Uh oh!

prodriguezdefino commented Dec 1, 2022

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

prodriguezdefino commented Nov 18, 2022 •

edited

Loading

prodriguezdefino Dec 1, 2022 •

edited

Loading

Abacn Dec 2, 2022 •

edited

Loading

prodriguezdefino Dec 1, 2022 •

edited

Loading