[BEAM-11408] Integrate BigQuery sink streaming inserts with GroupIntoBatches

[nehsyc]

Use GroupIntoBatches.WithShardedKey API to group and batch write before streaming to BigQuery service. Currently batching is done best-effort on bundle finalization.

This PR

• adds an option to BigQueryOptions to toggle between the existing and new implementation;
• extracts the shared code between the old and new implementation to a class BatchedStreamingWrite and provides an option to choose the implementation.

---

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).
• Format the pull request title like [BEAM-XXX] Fixes bug in ApproximateQuantiles, where you replace BEAM-XXX with the appropriate JIRA issue, if applicable. This will automatically link the pull request to the issue.
• 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.

Post-Commit Tests Status (on master branch)

Lang	SDK	Dataflow	Flink	Samza	Spark	Twister2
Go		---		---		---
Java
Python				---		---
XLang				---		---

Pre-Commit Tests Status (on master branch)

---	Java	Python	Go	Website	Whitespace	Typescript
Non-portable
Portable	---		---	---	---	---

See .test-infra/jenkins/README for trigger phrase, status and link of all Jenkins jobs.

GitHub Actions Tests Status (on master branch)

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

[nehsyc]

R: @reuvenlax @chamikaramj

[nehsyc]

Not sure about this limit. What would be a proper value? Should we make it configurable?

[pabloem]

to be hones, I am not sure what's a good duration either. I think this is acceptable for now, until we find out more. Thoughts?

[nehsyc]

Yeah sounds reasonable to proceed with this for now.

[reuvenlax]

We are relying on the fact that the GroupIntoBatches produces stable output. Really we should tag this with RequiresStableInput. Can you find out if this is safe to do in Dataflow?

[nehsyc]

There is a transform override in Dataflow to add a preceding Reshuffle to DoFns marked with RequiresStableInput.

beam/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/RequiresStableInputParDoOverrides.java

Line 73 in 74ec609

.apply("Materialize input", Reshuffle.viaRandomKey())

The override is disabled though. So I guess currently Dataflow does nothing for this tag.

beam/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowRunner.java

Line 601 in df74d74

/* TODO[Beam-4684]: Support @RequiresStableInput on Dataflow in a more intelligent way

Also my understanding is that dding a Reshuffle before GroupIntoBatches will introduce an extra shuffle as Reshuffle is essentially a GBK + value expansion in Dataflow.

[nehsyc]

@reuvenlax I also have changes for FILE_LOADS ready in my local branch. If it is ok for you, I can merge those into this PR. Otherwise I will send a follow-up PR.

[nehsyc]

R: @pabloem

[nehsyc]

R: @pabloem

Hey Pablo, let me know if it is ok for you to include the changes in FILE_LOADS in this PR as well. If so, I will push a new commit.

[pabloem]

Run Java PostCommit

[pabloem]

Run Java PreCommit

[pabloem]

There are a couple checkstyle warnings on Precommit:

[ant:checkstyle] [ERROR] /home/jenkins/jenkins-slave/workspace/beam_PreCommit_Java_Phrase/src/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BatchedStreamingWrite.java:85:24: Name 'LOG' must match pattern '^[a-z][a-zA-Z0-9]*$'. [MemberName] |  
  | [ant:checkstyle] [ERROR] /home/jenkins/jenkins-slave/workspace/beam_PreCommit_Java_Phrase/src/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BatchedStreamingWrite.java:279:28: Name 'BATCH_MAX_BUFFERING_DURATION' must match pattern '^[a-z][a-zA-Z0-9]*$'. [MemberName]

These mean that the variables should be static to have CAPITALIZED_NAMES, or should be named with camelCase if they are not static. Can you fix that?

Also, it seems there are some merge conflicts. Can you fix that as well?

The last thing to figure out is how to address Reuven's comment regarding stable input

[pabloem]

Run Java PostCommit

[pabloem]

Run Java_Examples_Dataflow_Java11 PreCommit

[nehsyc]

There are a couple checkstyle warnings on Precommit:

[ant:checkstyle] [ERROR] /home/jenkins/jenkins-slave/workspace/beam_PreCommit_Java_Phrase/src/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BatchedStreamingWrite.java:85:24: Name 'LOG' must match pattern '^[a-z][a-zA-Z0-9]*$'. [MemberName] |  
  | [ant:checkstyle] [ERROR] /home/jenkins/jenkins-slave/workspace/beam_PreCommit_Java_Phrase/src/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BatchedStreamingWrite.java:279:28: Name 'BATCH_MAX_BUFFERING_DURATION' must match pattern '^[a-z][a-zA-Z0-9]*$'. [MemberName]

These mean that the variables should be static to have CAPITALIZED_NAMES, or should be named with camelCase if they are not static. Can you fix that?

Thanks for pointing out! Pushed a commit to fix those.

Also, it seems there are some merge conflicts. Can you fix that as well?

Done.

The last thing to figure out is how to address Reuven's comment regarding stable input

Based on my understanding Dataflow currently doesn't do anything for @RequiresStableInputs so it may be considered as safe for now but if we add naive support like a Reshuffle it would be adding duplicated shuffles. How about adding a TODO here so we don't forget?

[pabloem]

There seems to be an issue with a try/catch: https://ci-beam.apache.org/job/beam_PreCommit_Java_Examples_Dataflow_Commit/12361/console

[pabloem]

Run Java PostCommit

[pabloem]

Run Java PreCommit

[pabloem]

Postcommit from previous commit: https://ci-beam.apache.org/job/beam_PostCommit_Java_PR/558/

[pabloem]

Thanks @nehsyc !