Merge pull request #30059 [YAML] Require numStreams for unbounded Big…

…QueryStorageWrite Require numStreams for unbounded BigQueryStorageWriteApiSchemaTransform xlang transform. This is to work around #30058
apache · Jan 24, 2024 · 11646ca · 11646ca
2 parents 0eefb79 + 8f3887b
commit 11646ca
Show file tree

Hide file tree

Showing 2 changed files with 10 additions and 0 deletions.
diff --git a/...he/beam/sdk/io/gcp/bigquery/providers/BigQueryStorageWriteApiSchemaTransformProvider.java b/...he/beam/sdk/io/gcp/bigquery/providers/BigQueryStorageWriteApiSchemaTransformProvider.java
@@ -349,6 +349,14 @@ public PCollectionRowTuple expand(PCollectionRowTuple input) {
         Long triggeringFrequency = configuration.getTriggeringFrequencySeconds();
         Boolean autoSharding = configuration.getAutoSharding();
         int numStreams = configuration.getNumStreams() == null ? 0 : configuration.getNumStreams();
+
+        // TODO(https://github.com/apache/beam/issues/30058): remove once Dataflow supports multiple
+        // DoFn's per fused step.
+        if (numStreams < 1) {
+          throw new IllegalStateException(
+              "numStreams must be set to a positive integer when input data is unbounded.");
+        }
+
         boolean useAtLeastOnceSemantics =
             configuration.getUseAtLeastOnceSemantics() == null
                 ? false

diff --git a/sdks/python/apache_beam/yaml/standard_io.yaml b/sdks/python/apache_beam/yaml/standard_io.yaml
@@ -38,6 +38,8 @@
         create_disposition: 'createDisposition'
         write_disposition: 'writeDisposition'
         error_handling: 'errorHandling'
+        # TODO(https://github.com/apache/beam/issues/30058): Required until autosharding support is fixed
+        num_streams: 'numStreams'
     underlying_provider:
       type: beamJar
       transforms: