You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by "Naireen (via GitHub)" <gi...@apache.org> on 2023/10/31 20:56:09 UTC
[PR] Test sink name [beam]
Naireen opened a new pull request, #29230:
URL: https://github.com/apache/beam/pull/29230
Add tests to check the name of the transform
------------------------
Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:
- [ ] 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](https://www.apache.org/licenses/icla.pdf).
See the [Contributor Guide](https://beam.apache.org/contribute) for more tips on [how to make review process smoother](https://github.com/apache/beam/blob/master/CONTRIBUTING.md#make-the-reviewers-job-easier).
To check the build health, please visit [https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md](https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md)
GitHub Actions Tests Status (on master branch)
------------------------------------------------------------------------------------------------
[](https://github.com/apache/beam/actions?query=workflow%3A%22Build+python+source+distribution+and+wheels%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Python+Tests%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Java+Tests%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Go+tests%22+branch%3Amaster+event%3Aschedule)
See [CI.md](https://github.com/apache/beam/blob/master/CI.md) for more information about GitHub Actions CI or the [workflows README](https://github.com/apache/beam/blob/master/.github/workflows/README.md) to see a list of phrases to trigger workflows.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Test sink name [beam]
Posted by "y1chi (via GitHub)" <gi...@apache.org>.
y1chi commented on code in PR #29230:
URL: https://github.com/apache/beam/pull/29230#discussion_r1378212715
##########
sdks/java/core/src/test/java/org/apache/beam/sdk/io/WriteFilesTest.java:
##########
@@ -477,6 +480,39 @@ public void testBuildWrite() {
assertThat(write.getComputeNumShards(), equalTo(originalSharding));
}
+ @Test
+ public void testWriteTransformName() {
Review Comment:
Anywhere in runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow we can put these tests? It is very specific for dataflow that I don't feel like it is good idea to put it under sdk.
Also can we create a github issue as a TODO for the tests for a more generic solution for identifying these transforms when submitting to dataflow via v1b3 API. We should remove these tests after we have a more safe solution.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Test sink name [beam]
Posted by "y1chi (via GitHub)" <gi...@apache.org>.
y1chi commented on code in PR #29230:
URL: https://github.com/apache/beam/pull/29230#discussion_r1384063689
##########
runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/DataflowPipelineTranslatorTest.java:
##########
@@ -229,6 +238,126 @@ private static DataflowPipelineOptions buildPipelineOptions() throws IOException
return options;
}
+ // Test that the transform names are what we expect them to be
+ @Test
+ public void testStorageWriteApiTransformNames() throws IOException, Exception {
+
+ DataflowPipelineOptions options = buildPipelineOptions();
+ options.setRunner(DataflowRunner.class);
+ Pipeline p = Pipeline.create(options);
+
+ p.traverseTopologically(new RecordingPipelineVisitor());
+ SdkComponents sdkComponents = createSdkComponents(options);
+
+ List<String> inputs = Lists.newArrayList();
Review Comment:
you could probably just inline some input elements with Create.of(), since the input is not important we can make it much simpler.
##########
runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/DataflowPipelineTranslatorTest.java:
##########
@@ -229,6 +238,126 @@ private static DataflowPipelineOptions buildPipelineOptions() throws IOException
return options;
}
+ // Test that the transform names are what we expect them to be
Review Comment:
Can we make the comment more clear what is expected and why. Also please add a TODO comment with github issue link for more stable solution.
##########
runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/DataflowPipelineTranslatorTest.java:
##########
@@ -229,6 +238,126 @@ private static DataflowPipelineOptions buildPipelineOptions() throws IOException
return options;
}
+ // Test that the transform names are what we expect them to be
+ @Test
+ public void testStorageWriteApiTransformNames() throws IOException, Exception {
+
+ DataflowPipelineOptions options = buildPipelineOptions();
+ options.setRunner(DataflowRunner.class);
+ Pipeline p = Pipeline.create(options);
+
+ p.traverseTopologically(new RecordingPipelineVisitor());
+ SdkComponents sdkComponents = createSdkComponents(options);
+
+ List<String> inputs = Lists.newArrayList();
+ for (int i = 0; i < 100; ++i) {
+ inputs.add("mambo_number_" + i);
+ }
+
+ BigQueryIO.Write<String> writeTransform =
+ BigQueryIO.<String>write()
+ .withFormatFunction(
+ (SerializableFunction<String, TableRow>)
+ input1 -> new TableRow().set("description", input1))
+ .withMethod(BigQueryIO.Write.Method.STORAGE_WRITE_API)
+ .withCreateDisposition(CreateDisposition.CREATE_IF_NEEDED)
+ .withTriggeringFrequency(org.joda.time.Duration.standardSeconds(5))
+ .to(
+ new TableReference()
+ .setProjectId("project")
+ .setDatasetId("dataset")
+ .setTableId("table"))
+ .withSchema(
+ new TableSchema()
+ .setFields(
+ new ArrayList<>(
+ ImmutableList.of(
+ new TableFieldSchema().setName("description").setType("STRING")))));
+
+ p.apply(Create.of(inputs).withCoder(StringUtf8Coder.of()))
+ .setIsBoundedInternal(IsBounded.UNBOUNDED)
+ .apply("StorageWriteApi", writeTransform);
+
+ RunnerApi.Pipeline pipelineProto = PipelineTranslation.toProto(p, sdkComponents, true);
+ DataflowPipelineTranslator t =
+ DataflowPipelineTranslator.fromOptions(
+ PipelineOptionsFactory.as(DataflowPipelineOptions.class));
+
+ JobSpecification jobSpecification =
+ t.translate(
+ p,
+ pipelineProto,
+ sdkComponents,
+ DataflowRunner.fromOptions(options),
+ Collections.emptyList());
+ assertAllStepOutputsHaveUniqueIds(jobSpecification.getJob());
Review Comment:
Is this check relevant to what we are testing?
##########
runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/DataflowPipelineTranslatorTest.java:
##########
@@ -229,6 +238,126 @@ private static DataflowPipelineOptions buildPipelineOptions() throws IOException
return options;
}
+ // Test that the transform names are what we expect them to be
+ @Test
+ public void testStorageWriteApiTransformNames() throws IOException, Exception {
+
+ DataflowPipelineOptions options = buildPipelineOptions();
+ options.setRunner(DataflowRunner.class);
+ Pipeline p = Pipeline.create(options);
+
+ p.traverseTopologically(new RecordingPipelineVisitor());
+ SdkComponents sdkComponents = createSdkComponents(options);
+
+ List<String> inputs = Lists.newArrayList();
+ for (int i = 0; i < 100; ++i) {
+ inputs.add("mambo_number_" + i);
+ }
+
+ BigQueryIO.Write<String> writeTransform =
+ BigQueryIO.<String>write()
+ .withFormatFunction(
+ (SerializableFunction<String, TableRow>)
+ input1 -> new TableRow().set("description", input1))
+ .withMethod(BigQueryIO.Write.Method.STORAGE_WRITE_API)
+ .withCreateDisposition(CreateDisposition.CREATE_IF_NEEDED)
+ .withTriggeringFrequency(org.joda.time.Duration.standardSeconds(5))
+ .to(
+ new TableReference()
+ .setProjectId("project")
+ .setDatasetId("dataset")
+ .setTableId("table"))
+ .withSchema(
+ new TableSchema()
+ .setFields(
+ new ArrayList<>(
+ ImmutableList.of(
+ new TableFieldSchema().setName("description").setType("STRING")))));
+
+ p.apply(Create.of(inputs).withCoder(StringUtf8Coder.of()))
+ .setIsBoundedInternal(IsBounded.UNBOUNDED)
+ .apply("StorageWriteApi", writeTransform);
+
+ RunnerApi.Pipeline pipelineProto = PipelineTranslation.toProto(p, sdkComponents, true);
+ DataflowPipelineTranslator t =
+ DataflowPipelineTranslator.fromOptions(
+ PipelineOptionsFactory.as(DataflowPipelineOptions.class));
+
+ JobSpecification jobSpecification =
+ t.translate(
+ p,
+ pipelineProto,
+ sdkComponents,
+ DataflowRunner.fromOptions(options),
+ Collections.emptyList());
+ assertAllStepOutputsHaveUniqueIds(jobSpecification.getJob());
+
+ // Assert that at least one of the steps added was the StorageApiLoads step. This is added to
+ // ensure the name doesn't change.
+ boolean foundStep = false;
+ for (Step step : jobSpecification.getJob().getSteps()) {
+ if (getString(step.getProperties(), PropertyNames.USER_NAME).contains("/StorageApiLoads")) {
Review Comment:
I think StorageWriteApi should also be included in this step name? We can be more specific that StorageWriteApi step contains "/StorageApiLoads"
##########
runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/DataflowPipelineTranslatorTest.java:
##########
@@ -229,6 +238,126 @@ private static DataflowPipelineOptions buildPipelineOptions() throws IOException
return options;
}
+ // Test that the transform names are what we expect them to be
+ @Test
+ public void testStorageWriteApiTransformNames() throws IOException, Exception {
+
+ DataflowPipelineOptions options = buildPipelineOptions();
+ options.setRunner(DataflowRunner.class);
+ Pipeline p = Pipeline.create(options);
+
+ p.traverseTopologically(new RecordingPipelineVisitor());
+ SdkComponents sdkComponents = createSdkComponents(options);
+
+ List<String> inputs = Lists.newArrayList();
+ for (int i = 0; i < 100; ++i) {
+ inputs.add("mambo_number_" + i);
+ }
+
+ BigQueryIO.Write<String> writeTransform =
+ BigQueryIO.<String>write()
+ .withFormatFunction(
+ (SerializableFunction<String, TableRow>)
+ input1 -> new TableRow().set("description", input1))
+ .withMethod(BigQueryIO.Write.Method.STORAGE_WRITE_API)
+ .withCreateDisposition(CreateDisposition.CREATE_IF_NEEDED)
+ .withTriggeringFrequency(org.joda.time.Duration.standardSeconds(5))
+ .to(
+ new TableReference()
+ .setProjectId("project")
+ .setDatasetId("dataset")
+ .setTableId("table"))
+ .withSchema(
+ new TableSchema()
+ .setFields(
+ new ArrayList<>(
+ ImmutableList.of(
+ new TableFieldSchema().setName("description").setType("STRING")))));
+
+ p.apply(Create.of(inputs).withCoder(StringUtf8Coder.of()))
Review Comment:
can we combine these two tests (StorageApiLoads and WriteAutoSharededBundlesToTempFiles) into 1. We can just use different branches.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Test sink name [beam]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #29230:
URL: https://github.com/apache/beam/pull/29230#issuecomment-1788155259
Checks are failing. Will not request review until checks are succeeding. If you'd like to override that behavior, comment `assign set of reviewers`
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Test sink name [beam]
Posted by "y1chi (via GitHub)" <gi...@apache.org>.
y1chi commented on code in PR #29230:
URL: https://github.com/apache/beam/pull/29230#discussion_r1378221242
##########
sdks/java/core/src/test/java/org/apache/beam/sdk/io/WriteFilesTest.java:
##########
@@ -477,6 +480,39 @@ public void testBuildWrite() {
assertThat(write.getComputeNumShards(), equalTo(originalSharding));
}
+ @Test
+ public void testWriteTransformName() {
+ PipelineOptions options = TestPipeline.testingPipelineOptions();
+ Pipeline pipeline = Pipeline.create(options);
+
+ SimpleSink<Void> sink = makeSimpleSink();
+ WriteFiles<String, ?, String> write =
+ WriteFiles.to(sink).withNumShards(0).withWindowedWrites().withRunnerDeterminedSharding();
+
+ List<String> inputs = Lists.newArrayList();
+ for (int i = 0; i < 100; ++i) {
+ inputs.add("mambo_number_" + i);
+ }
+
+ // Prepare timestamps for the elements.
+ List<Long> timestamps = new ArrayList<>();
+ for (long i = 0; i < inputs.size(); i++) {
+ timestamps.add(i + 1);
+ }
+
+ assertThat((SimpleSink<Void>) write.getSink(), is(sink));
+ assertThat(write.getComputeNumShards(), nullValue());
+
+ pipeline
+ .apply(Create.timestamped(inputs, timestamps).withCoder(StringUtf8Coder.of()))
+ .setIsBoundedInternal(IsBounded.UNBOUNDED)
+ .apply(Window.into(FixedWindows.of(Duration.millis(1))))
+ .apply(write);
+ // This should not change, this ensures that the transform name is what the dataflow runner
+ // expects.
+ logged.verifyDebug("WriteAutoShardedBundlesToTempFiles");
Review Comment:
Can we test more directly? For example we may be able to get the step name from translated job specification https://github.com/apache/beam/blob/d329a7e9b85d729a474395abfe1975c0f3649550/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowPipelineTranslator.java#L199
just in case the name is somehow changed or removed after translation
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Test sink name [beam]
Posted by "Naireen (via GitHub)" <gi...@apache.org>.
Naireen commented on code in PR #29230:
URL: https://github.com/apache/beam/pull/29230#discussion_r1385603743
##########
runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/DataflowPipelineTranslatorTest.java:
##########
@@ -229,6 +238,126 @@ private static DataflowPipelineOptions buildPipelineOptions() throws IOException
return options;
}
+ // Test that the transform names are what we expect them to be
Review Comment:
Updated, plus added link to issue.
##########
runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/DataflowPipelineTranslatorTest.java:
##########
@@ -229,6 +238,126 @@ private static DataflowPipelineOptions buildPipelineOptions() throws IOException
return options;
}
+ // Test that the transform names are what we expect them to be
+ @Test
+ public void testStorageWriteApiTransformNames() throws IOException, Exception {
+
+ DataflowPipelineOptions options = buildPipelineOptions();
+ options.setRunner(DataflowRunner.class);
+ Pipeline p = Pipeline.create(options);
+
+ p.traverseTopologically(new RecordingPipelineVisitor());
+ SdkComponents sdkComponents = createSdkComponents(options);
+
+ List<String> inputs = Lists.newArrayList();
+ for (int i = 0; i < 100; ++i) {
+ inputs.add("mambo_number_" + i);
+ }
+
+ BigQueryIO.Write<String> writeTransform =
+ BigQueryIO.<String>write()
+ .withFormatFunction(
+ (SerializableFunction<String, TableRow>)
+ input1 -> new TableRow().set("description", input1))
+ .withMethod(BigQueryIO.Write.Method.STORAGE_WRITE_API)
+ .withCreateDisposition(CreateDisposition.CREATE_IF_NEEDED)
+ .withTriggeringFrequency(org.joda.time.Duration.standardSeconds(5))
+ .to(
+ new TableReference()
+ .setProjectId("project")
+ .setDatasetId("dataset")
+ .setTableId("table"))
+ .withSchema(
+ new TableSchema()
+ .setFields(
+ new ArrayList<>(
+ ImmutableList.of(
+ new TableFieldSchema().setName("description").setType("STRING")))));
+
+ p.apply(Create.of(inputs).withCoder(StringUtf8Coder.of()))
+ .setIsBoundedInternal(IsBounded.UNBOUNDED)
+ .apply("StorageWriteApi", writeTransform);
+
+ RunnerApi.Pipeline pipelineProto = PipelineTranslation.toProto(p, sdkComponents, true);
+ DataflowPipelineTranslator t =
+ DataflowPipelineTranslator.fromOptions(
+ PipelineOptionsFactory.as(DataflowPipelineOptions.class));
+
+ JobSpecification jobSpecification =
+ t.translate(
+ p,
+ pipelineProto,
+ sdkComponents,
+ DataflowRunner.fromOptions(options),
+ Collections.emptyList());
+ assertAllStepOutputsHaveUniqueIds(jobSpecification.getJob());
+
+ // Assert that at least one of the steps added was the StorageApiLoads step. This is added to
+ // ensure the name doesn't change.
+ boolean foundStep = false;
+ for (Step step : jobSpecification.getJob().getSteps()) {
+ if (getString(step.getProperties(), PropertyNames.USER_NAME).contains("/StorageApiLoads")) {
Review Comment:
Done.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Test sink name [beam]
Posted by "Naireen (via GitHub)" <gi...@apache.org>.
Naireen commented on code in PR #29230:
URL: https://github.com/apache/beam/pull/29230#discussion_r1387055705
##########
runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/DataflowPipelineTranslatorTest.java:
##########
@@ -229,6 +237,115 @@ private static DataflowPipelineOptions buildPipelineOptions() throws IOException
return options;
}
+ // Test that the transform names for Storage Write API for streaming pipelines are what we expect
+ // them to be. This is required since the Windmill backend expects the step to contain that name.
+ // For a more stable solution, we use use urn, but that is not currently used in the legacy java
+ // worker.
+ // TODO:(https://github.com/apache/beam/issues/29338) Pass in URN information to GCP Runner.
+ @Test
+ public void testStorageWriteApiTransformNames() throws IOException, Exception {
+
+ DataflowPipelineOptions options = buildPipelineOptions();
+ options.setRunner(DataflowRunner.class);
+ Pipeline p = Pipeline.create(options);
+
+ p.traverseTopologically(new RecordingPipelineVisitor());
+ SdkComponents sdkComponents = createSdkComponents(options);
+
+ BigQueryIO.Write<String> writeTransform =
+ BigQueryIO.<String>write()
+ .withFormatFunction(
+ (SerializableFunction<String, TableRow>)
+ input1 -> new TableRow().set("description", input1))
+ .withMethod(BigQueryIO.Write.Method.STORAGE_WRITE_API)
+ .withCreateDisposition(CreateDisposition.CREATE_IF_NEEDED)
+ .withTriggeringFrequency(org.joda.time.Duration.standardSeconds(5))
+ .to(
+ new TableReference()
+ .setProjectId("project")
+ .setDatasetId("dataset")
+ .setTableId("table"))
+ .withSchema(
+ new TableSchema()
+ .setFields(
+ new ArrayList<>(
+ ImmutableList.of(
+ new TableFieldSchema().setName("description").setType("STRING")))));
+
+ p.apply(Create.of("1", "2", "3", "4").withCoder(StringUtf8Coder.of()))
+ .setIsBoundedInternal(IsBounded.UNBOUNDED)
+ .apply("StorageWriteApi", writeTransform);
+
+ RunnerApi.Pipeline pipelineProto = PipelineTranslation.toProto(p, sdkComponents, true);
+ DataflowPipelineTranslator t =
+ DataflowPipelineTranslator.fromOptions(
+ PipelineOptionsFactory.as(DataflowPipelineOptions.class));
+
+ JobSpecification jobSpecification =
+ t.translate(
+ p,
+ pipelineProto,
+ sdkComponents,
+ DataflowRunner.fromOptions(options),
+ Collections.emptyList());
+
+ boolean foundStep = false;
+ for (Step step : jobSpecification.getJob().getSteps()) {
+ if (getString(step.getProperties(), PropertyNames.USER_NAME)
+ .contains("StorageWriteApi/StorageApiLoads")) {
+ foundStep = true;
+ }
+ }
+ assertTrue(foundStep);
+ }
+
+ // Test that the transform names added for TextIO writes with autosharding. This is required since
+ // the Windmill backend expects the file write autosharded step to contain that name.
+ // For a more stable solution, we use use urn, but that is not currently used in the legacy java
+ // worker.
+ // TODO:(https://github.com/apache/beam/issues/29338) Pass in URN information to GCP Runner.
+ @Test
+ public void testGCSWriteTransformNames() throws IOException, Exception {
+
+ DataflowPipelineOptions options = buildPipelineOptions();
+ options.setRunner(DataflowRunner.class);
+ Pipeline p = Pipeline.create(options);
+
+ p.traverseTopologically(new RecordingPipelineVisitor());
+ SdkComponents sdkComponents = createSdkComponents(options);
+
+ p.apply(Create.of("1", "2", "3", "4").withCoder(StringUtf8Coder.of()))
+ .setIsBoundedInternal(IsBounded.UNBOUNDED)
+ .apply(Window.into(FixedWindows.of(Duration.millis(1))))
+ .apply(
+ "WriteMyFile",
+ TextIO.write().to("gs://bucket/object").withWindowedWrites().withNumShards(0));
+
+ RunnerApi.Pipeline pipelineProto = PipelineTranslation.toProto(p, sdkComponents, true);
+ DataflowPipelineTranslator t =
+ DataflowPipelineTranslator.fromOptions(
+ PipelineOptionsFactory.as(DataflowPipelineOptions.class));
+
+ JobSpecification jobSpecification =
+ t.translate(
+ p,
+ pipelineProto,
+ sdkComponents,
+ DataflowRunner.fromOptions(options),
+ Collections.emptyList());
+
+ // Assert that at least one of the steps added was the autosharded write. This is added to
+ // ensure the name doesn't change.
+ boolean foundStep = false;
+ for (Step step : jobSpecification.getJob().getSteps()) {
+ if (getString(step.getProperties(), PropertyNames.USER_NAME)
+ .contains("/WriteAutoShardedBundlesToTempFiles")) {
Review Comment:
So its WriteMyFile/WriteFiles/FinalizeTempFileBundles/, I changed this to check for "WriteFiles/WriteAutoShardedBundlesToTempFile, since we don't care about what the user has set.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Test sink name [beam]
Posted by "y1chi (via GitHub)" <gi...@apache.org>.
y1chi commented on code in PR #29230:
URL: https://github.com/apache/beam/pull/29230#discussion_r1378221390
##########
sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIOWriteTest.java:
##########
@@ -1293,6 +1295,10 @@ private void storageWrite(boolean autoSharding) throws Exception {
p.apply(source).apply("WriteToBQ", write);
p.run().waitUntilFinish();
+ // This should not change, this ensures that the transform name is what the dataflow runner
+ // expects.
+ logged.verifyDebug("/StorageApiLoads");
Review Comment:
ditto.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Test sink name [beam]
Posted by "y1chi (via GitHub)" <gi...@apache.org>.
y1chi merged PR #29230:
URL: https://github.com/apache/beam/pull/29230
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Test sink name [beam]
Posted by "y1chi (via GitHub)" <gi...@apache.org>.
y1chi commented on code in PR #29230:
URL: https://github.com/apache/beam/pull/29230#discussion_r1378212715
##########
sdks/java/core/src/test/java/org/apache/beam/sdk/io/WriteFilesTest.java:
##########
@@ -477,6 +480,39 @@ public void testBuildWrite() {
assertThat(write.getComputeNumShards(), equalTo(originalSharding));
}
+ @Test
+ public void testWriteTransformName() {
Review Comment:
Anywhere in runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow we can put these tests, maybe DataflowPipelineTranslatorTest.java? It is very specific for dataflow that I don't feel like it is good idea to put it under sdk.
Also can we create a github issue as a TODO for the tests for a more generic solution for identifying these transforms when submitting to dataflow via v1b3 API. We should remove these tests after we have a more safe solution.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Test sink name [beam]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #29230:
URL: https://github.com/apache/beam/pull/29230#issuecomment-1789755997
Assigning reviewers. If you would like to opt out of this review, comment `assign to next reviewer`:
R: @chamikaramj added as fallback since no labels match configuration
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).
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Test sink name [beam]
Posted by "Naireen (via GitHub)" <gi...@apache.org>.
Naireen commented on code in PR #29230:
URL: https://github.com/apache/beam/pull/29230#discussion_r1385605168
##########
runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/DataflowPipelineTranslatorTest.java:
##########
@@ -229,6 +238,126 @@ private static DataflowPipelineOptions buildPipelineOptions() throws IOException
return options;
}
+ // Test that the transform names are what we expect them to be
+ @Test
+ public void testStorageWriteApiTransformNames() throws IOException, Exception {
+
+ DataflowPipelineOptions options = buildPipelineOptions();
+ options.setRunner(DataflowRunner.class);
+ Pipeline p = Pipeline.create(options);
+
+ p.traverseTopologically(new RecordingPipelineVisitor());
+ SdkComponents sdkComponents = createSdkComponents(options);
+
+ List<String> inputs = Lists.newArrayList();
Review Comment:
Done
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Test sink name [beam]
Posted by "Naireen (via GitHub)" <gi...@apache.org>.
Naireen commented on code in PR #29230:
URL: https://github.com/apache/beam/pull/29230#discussion_r1385604087
##########
runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/DataflowPipelineTranslatorTest.java:
##########
@@ -229,6 +238,126 @@ private static DataflowPipelineOptions buildPipelineOptions() throws IOException
return options;
}
+ // Test that the transform names are what we expect them to be
+ @Test
+ public void testStorageWriteApiTransformNames() throws IOException, Exception {
+
+ DataflowPipelineOptions options = buildPipelineOptions();
+ options.setRunner(DataflowRunner.class);
+ Pipeline p = Pipeline.create(options);
+
+ p.traverseTopologically(new RecordingPipelineVisitor());
+ SdkComponents sdkComponents = createSdkComponents(options);
+
+ List<String> inputs = Lists.newArrayList();
+ for (int i = 0; i < 100; ++i) {
+ inputs.add("mambo_number_" + i);
+ }
+
+ BigQueryIO.Write<String> writeTransform =
+ BigQueryIO.<String>write()
+ .withFormatFunction(
+ (SerializableFunction<String, TableRow>)
+ input1 -> new TableRow().set("description", input1))
+ .withMethod(BigQueryIO.Write.Method.STORAGE_WRITE_API)
+ .withCreateDisposition(CreateDisposition.CREATE_IF_NEEDED)
+ .withTriggeringFrequency(org.joda.time.Duration.standardSeconds(5))
+ .to(
+ new TableReference()
+ .setProjectId("project")
+ .setDatasetId("dataset")
+ .setTableId("table"))
+ .withSchema(
+ new TableSchema()
+ .setFields(
+ new ArrayList<>(
+ ImmutableList.of(
+ new TableFieldSchema().setName("description").setType("STRING")))));
+
+ p.apply(Create.of(inputs).withCoder(StringUtf8Coder.of()))
+ .setIsBoundedInternal(IsBounded.UNBOUNDED)
+ .apply("StorageWriteApi", writeTransform);
+
+ RunnerApi.Pipeline pipelineProto = PipelineTranslation.toProto(p, sdkComponents, true);
+ DataflowPipelineTranslator t =
+ DataflowPipelineTranslator.fromOptions(
+ PipelineOptionsFactory.as(DataflowPipelineOptions.class));
+
+ JobSpecification jobSpecification =
+ t.translate(
+ p,
+ pipelineProto,
+ sdkComponents,
+ DataflowRunner.fromOptions(options),
+ Collections.emptyList());
+ assertAllStepOutputsHaveUniqueIds(jobSpecification.getJob());
Review Comment:
No, so I removed it.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Test sink name [beam]
Posted by "Naireen (via GitHub)" <gi...@apache.org>.
Naireen commented on code in PR #29230:
URL: https://github.com/apache/beam/pull/29230#discussion_r1379339872
##########
sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIOWriteTest.java:
##########
@@ -1293,6 +1295,10 @@ private void storageWrite(boolean autoSharding) throws Exception {
p.apply(source).apply("WriteToBQ", write);
p.run().waitUntilFinish();
+ // This should not change, this ensures that the transform name is what the dataflow runner
+ // expects.
+ logged.verifyDebug("/StorageApiLoads");
Review Comment:
Done
##########
sdks/java/core/src/test/java/org/apache/beam/sdk/io/WriteFilesTest.java:
##########
@@ -477,6 +480,39 @@ public void testBuildWrite() {
assertThat(write.getComputeNumShards(), equalTo(originalSharding));
}
+ @Test
+ public void testWriteTransformName() {
+ PipelineOptions options = TestPipeline.testingPipelineOptions();
+ Pipeline pipeline = Pipeline.create(options);
+
+ SimpleSink<Void> sink = makeSimpleSink();
+ WriteFiles<String, ?, String> write =
+ WriteFiles.to(sink).withNumShards(0).withWindowedWrites().withRunnerDeterminedSharding();
+
+ List<String> inputs = Lists.newArrayList();
+ for (int i = 0; i < 100; ++i) {
+ inputs.add("mambo_number_" + i);
+ }
+
+ // Prepare timestamps for the elements.
+ List<Long> timestamps = new ArrayList<>();
+ for (long i = 0; i < inputs.size(); i++) {
+ timestamps.add(i + 1);
+ }
+
+ assertThat((SimpleSink<Void>) write.getSink(), is(sink));
+ assertThat(write.getComputeNumShards(), nullValue());
+
+ pipeline
+ .apply(Create.timestamped(inputs, timestamps).withCoder(StringUtf8Coder.of()))
+ .setIsBoundedInternal(IsBounded.UNBOUNDED)
+ .apply(Window.into(FixedWindows.of(Duration.millis(1))))
+ .apply(write);
+ // This should not change, this ensures that the transform name is what the dataflow runner
+ // expects.
+ logged.verifyDebug("WriteAutoShardedBundlesToTempFiles");
Review Comment:
Done.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Test sink name [beam]
Posted by "y1chi (via GitHub)" <gi...@apache.org>.
y1chi commented on code in PR #29230:
URL: https://github.com/apache/beam/pull/29230#discussion_r1385612930
##########
runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/DataflowPipelineTranslatorTest.java:
##########
@@ -229,6 +237,115 @@ private static DataflowPipelineOptions buildPipelineOptions() throws IOException
return options;
}
+ // Test that the transform names for Storage Write API for streaming pipelines are what we expect
+ // them to be. This is required since the Windmill backend expects the step to contain that name.
+ // For a more stable solution, we use use urn, but that is not currently used in the legacy java
Review Comment:
```suggestion
// For a more stable solution, we should use URN, but that is not currently used in the legacy java
```
##########
runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/DataflowPipelineTranslatorTest.java:
##########
@@ -229,6 +237,115 @@ private static DataflowPipelineOptions buildPipelineOptions() throws IOException
return options;
}
+ // Test that the transform names for Storage Write API for streaming pipelines are what we expect
+ // them to be. This is required since the Windmill backend expects the step to contain that name.
+ // For a more stable solution, we use use urn, but that is not currently used in the legacy java
+ // worker.
+ // TODO:(https://github.com/apache/beam/issues/29338) Pass in URN information to GCP Runner.
+ @Test
+ public void testStorageWriteApiTransformNames() throws IOException, Exception {
+
+ DataflowPipelineOptions options = buildPipelineOptions();
+ options.setRunner(DataflowRunner.class);
+ Pipeline p = Pipeline.create(options);
+
+ p.traverseTopologically(new RecordingPipelineVisitor());
+ SdkComponents sdkComponents = createSdkComponents(options);
+
+ BigQueryIO.Write<String> writeTransform =
+ BigQueryIO.<String>write()
+ .withFormatFunction(
+ (SerializableFunction<String, TableRow>)
+ input1 -> new TableRow().set("description", input1))
+ .withMethod(BigQueryIO.Write.Method.STORAGE_WRITE_API)
+ .withCreateDisposition(CreateDisposition.CREATE_IF_NEEDED)
+ .withTriggeringFrequency(org.joda.time.Duration.standardSeconds(5))
+ .to(
+ new TableReference()
+ .setProjectId("project")
+ .setDatasetId("dataset")
+ .setTableId("table"))
+ .withSchema(
+ new TableSchema()
+ .setFields(
+ new ArrayList<>(
+ ImmutableList.of(
+ new TableFieldSchema().setName("description").setType("STRING")))));
+
+ p.apply(Create.of("1", "2", "3", "4").withCoder(StringUtf8Coder.of()))
+ .setIsBoundedInternal(IsBounded.UNBOUNDED)
+ .apply("StorageWriteApi", writeTransform);
+
+ RunnerApi.Pipeline pipelineProto = PipelineTranslation.toProto(p, sdkComponents, true);
+ DataflowPipelineTranslator t =
+ DataflowPipelineTranslator.fromOptions(
+ PipelineOptionsFactory.as(DataflowPipelineOptions.class));
+
+ JobSpecification jobSpecification =
+ t.translate(
+ p,
+ pipelineProto,
+ sdkComponents,
+ DataflowRunner.fromOptions(options),
+ Collections.emptyList());
+
+ boolean foundStep = false;
+ for (Step step : jobSpecification.getJob().getSteps()) {
+ if (getString(step.getProperties(), PropertyNames.USER_NAME)
+ .contains("StorageWriteApi/StorageApiLoads")) {
+ foundStep = true;
+ }
+ }
+ assertTrue(foundStep);
+ }
+
+ // Test that the transform names added for TextIO writes with autosharding. This is required since
+ // the Windmill backend expects the file write autosharded step to contain that name.
+ // For a more stable solution, we use use urn, but that is not currently used in the legacy java
Review Comment:
```suggestion
// For a more stable solution, we should use URN, but that is not currently used in the legacy java
```
##########
runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/DataflowPipelineTranslatorTest.java:
##########
@@ -229,6 +237,115 @@ private static DataflowPipelineOptions buildPipelineOptions() throws IOException
return options;
}
+ // Test that the transform names for Storage Write API for streaming pipelines are what we expect
+ // them to be. This is required since the Windmill backend expects the step to contain that name.
+ // For a more stable solution, we use use urn, but that is not currently used in the legacy java
+ // worker.
+ // TODO:(https://github.com/apache/beam/issues/29338) Pass in URN information to GCP Runner.
+ @Test
+ public void testStorageWriteApiTransformNames() throws IOException, Exception {
+
+ DataflowPipelineOptions options = buildPipelineOptions();
+ options.setRunner(DataflowRunner.class);
+ Pipeline p = Pipeline.create(options);
+
+ p.traverseTopologically(new RecordingPipelineVisitor());
+ SdkComponents sdkComponents = createSdkComponents(options);
+
+ BigQueryIO.Write<String> writeTransform =
+ BigQueryIO.<String>write()
+ .withFormatFunction(
+ (SerializableFunction<String, TableRow>)
+ input1 -> new TableRow().set("description", input1))
+ .withMethod(BigQueryIO.Write.Method.STORAGE_WRITE_API)
+ .withCreateDisposition(CreateDisposition.CREATE_IF_NEEDED)
+ .withTriggeringFrequency(org.joda.time.Duration.standardSeconds(5))
+ .to(
+ new TableReference()
+ .setProjectId("project")
+ .setDatasetId("dataset")
+ .setTableId("table"))
+ .withSchema(
+ new TableSchema()
+ .setFields(
+ new ArrayList<>(
+ ImmutableList.of(
+ new TableFieldSchema().setName("description").setType("STRING")))));
+
+ p.apply(Create.of("1", "2", "3", "4").withCoder(StringUtf8Coder.of()))
+ .setIsBoundedInternal(IsBounded.UNBOUNDED)
+ .apply("StorageWriteApi", writeTransform);
+
+ RunnerApi.Pipeline pipelineProto = PipelineTranslation.toProto(p, sdkComponents, true);
+ DataflowPipelineTranslator t =
+ DataflowPipelineTranslator.fromOptions(
+ PipelineOptionsFactory.as(DataflowPipelineOptions.class));
+
+ JobSpecification jobSpecification =
+ t.translate(
+ p,
+ pipelineProto,
+ sdkComponents,
+ DataflowRunner.fromOptions(options),
+ Collections.emptyList());
+
+ boolean foundStep = false;
+ for (Step step : jobSpecification.getJob().getSteps()) {
+ if (getString(step.getProperties(), PropertyNames.USER_NAME)
+ .contains("StorageWriteApi/StorageApiLoads")) {
+ foundStep = true;
+ }
+ }
+ assertTrue(foundStep);
+ }
+
+ // Test that the transform names added for TextIO writes with autosharding. This is required since
+ // the Windmill backend expects the file write autosharded step to contain that name.
+ // For a more stable solution, we use use urn, but that is not currently used in the legacy java
+ // worker.
+ // TODO:(https://github.com/apache/beam/issues/29338) Pass in URN information to GCP Runner.
+ @Test
+ public void testGCSWriteTransformNames() throws IOException, Exception {
+
+ DataflowPipelineOptions options = buildPipelineOptions();
+ options.setRunner(DataflowRunner.class);
+ Pipeline p = Pipeline.create(options);
+
+ p.traverseTopologically(new RecordingPipelineVisitor());
+ SdkComponents sdkComponents = createSdkComponents(options);
+
+ p.apply(Create.of("1", "2", "3", "4").withCoder(StringUtf8Coder.of()))
+ .setIsBoundedInternal(IsBounded.UNBOUNDED)
+ .apply(Window.into(FixedWindows.of(Duration.millis(1))))
+ .apply(
+ "WriteMyFile",
+ TextIO.write().to("gs://bucket/object").withWindowedWrites().withNumShards(0));
+
+ RunnerApi.Pipeline pipelineProto = PipelineTranslation.toProto(p, sdkComponents, true);
+ DataflowPipelineTranslator t =
+ DataflowPipelineTranslator.fromOptions(
+ PipelineOptionsFactory.as(DataflowPipelineOptions.class));
+
+ JobSpecification jobSpecification =
+ t.translate(
+ p,
+ pipelineProto,
+ sdkComponents,
+ DataflowRunner.fromOptions(options),
+ Collections.emptyList());
+
+ // Assert that at least one of the steps added was the autosharded write. This is added to
+ // ensure the name doesn't change.
+ boolean foundStep = false;
+ for (Step step : jobSpecification.getJob().getSteps()) {
+ if (getString(step.getProperties(), PropertyNames.USER_NAME)
+ .contains("/WriteAutoShardedBundlesToTempFiles")) {
Review Comment:
nit:WriteMyFile/WriteAutoShardedBundlesToTempFiles?
##########
runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/DataflowPipelineTranslatorTest.java:
##########
@@ -229,6 +237,115 @@ private static DataflowPipelineOptions buildPipelineOptions() throws IOException
return options;
}
+ // Test that the transform names for Storage Write API for streaming pipelines are what we expect
+ // them to be. This is required since the Windmill backend expects the step to contain that name.
+ // For a more stable solution, we use use urn, but that is not currently used in the legacy java
+ // worker.
+ // TODO:(https://github.com/apache/beam/issues/29338) Pass in URN information to GCP Runner.
+ @Test
+ public void testStorageWriteApiTransformNames() throws IOException, Exception {
+
+ DataflowPipelineOptions options = buildPipelineOptions();
+ options.setRunner(DataflowRunner.class);
+ Pipeline p = Pipeline.create(options);
+
+ p.traverseTopologically(new RecordingPipelineVisitor());
+ SdkComponents sdkComponents = createSdkComponents(options);
+
+ BigQueryIO.Write<String> writeTransform =
+ BigQueryIO.<String>write()
+ .withFormatFunction(
+ (SerializableFunction<String, TableRow>)
+ input1 -> new TableRow().set("description", input1))
+ .withMethod(BigQueryIO.Write.Method.STORAGE_WRITE_API)
+ .withCreateDisposition(CreateDisposition.CREATE_IF_NEEDED)
+ .withTriggeringFrequency(org.joda.time.Duration.standardSeconds(5))
+ .to(
+ new TableReference()
+ .setProjectId("project")
+ .setDatasetId("dataset")
+ .setTableId("table"))
+ .withSchema(
+ new TableSchema()
+ .setFields(
+ new ArrayList<>(
+ ImmutableList.of(
+ new TableFieldSchema().setName("description").setType("STRING")))));
+
+ p.apply(Create.of("1", "2", "3", "4").withCoder(StringUtf8Coder.of()))
+ .setIsBoundedInternal(IsBounded.UNBOUNDED)
+ .apply("StorageWriteApi", writeTransform);
+
+ RunnerApi.Pipeline pipelineProto = PipelineTranslation.toProto(p, sdkComponents, true);
+ DataflowPipelineTranslator t =
+ DataflowPipelineTranslator.fromOptions(
+ PipelineOptionsFactory.as(DataflowPipelineOptions.class));
+
+ JobSpecification jobSpecification =
+ t.translate(
+ p,
+ pipelineProto,
+ sdkComponents,
+ DataflowRunner.fromOptions(options),
+ Collections.emptyList());
+
+ boolean foundStep = false;
+ for (Step step : jobSpecification.getJob().getSteps()) {
+ if (getString(step.getProperties(), PropertyNames.USER_NAME)
+ .contains("StorageWriteApi/StorageApiLoads")) {
+ foundStep = true;
+ }
+ }
+ assertTrue(foundStep);
+ }
+
+ // Test that the transform names added for TextIO writes with autosharding. This is required since
+ // the Windmill backend expects the file write autosharded step to contain that name.
+ // For a more stable solution, we use use urn, but that is not currently used in the legacy java
+ // worker.
+ // TODO:(https://github.com/apache/beam/issues/29338) Pass in URN information to GCP Runner.
Review Comment:
```suggestion
// TODO:(https://github.com/apache/beam/issues/29338) Pass in URN information to Dataflow Runner.
```
##########
runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/DataflowPipelineTranslatorTest.java:
##########
@@ -229,6 +237,115 @@ private static DataflowPipelineOptions buildPipelineOptions() throws IOException
return options;
}
+ // Test that the transform names for Storage Write API for streaming pipelines are what we expect
+ // them to be. This is required since the Windmill backend expects the step to contain that name.
+ // For a more stable solution, we use use urn, but that is not currently used in the legacy java
+ // worker.
+ // TODO:(https://github.com/apache/beam/issues/29338) Pass in URN information to GCP Runner.
Review Comment:
```suggestion
// TODO:(https://github.com/apache/beam/issues/29338) Pass in URN information to Dataflow Runner.
```
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Test sink name [beam]
Posted by "Naireen (via GitHub)" <gi...@apache.org>.
Naireen commented on code in PR #29230:
URL: https://github.com/apache/beam/pull/29230#discussion_r1385604813
##########
runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/DataflowPipelineTranslatorTest.java:
##########
@@ -229,6 +238,126 @@ private static DataflowPipelineOptions buildPipelineOptions() throws IOException
return options;
}
+ // Test that the transform names are what we expect them to be
+ @Test
+ public void testStorageWriteApiTransformNames() throws IOException, Exception {
+
+ DataflowPipelineOptions options = buildPipelineOptions();
+ options.setRunner(DataflowRunner.class);
+ Pipeline p = Pipeline.create(options);
+
+ p.traverseTopologically(new RecordingPipelineVisitor());
+ SdkComponents sdkComponents = createSdkComponents(options);
+
+ List<String> inputs = Lists.newArrayList();
+ for (int i = 0; i < 100; ++i) {
+ inputs.add("mambo_number_" + i);
+ }
+
+ BigQueryIO.Write<String> writeTransform =
+ BigQueryIO.<String>write()
+ .withFormatFunction(
+ (SerializableFunction<String, TableRow>)
+ input1 -> new TableRow().set("description", input1))
+ .withMethod(BigQueryIO.Write.Method.STORAGE_WRITE_API)
+ .withCreateDisposition(CreateDisposition.CREATE_IF_NEEDED)
+ .withTriggeringFrequency(org.joda.time.Duration.standardSeconds(5))
+ .to(
+ new TableReference()
+ .setProjectId("project")
+ .setDatasetId("dataset")
+ .setTableId("table"))
+ .withSchema(
+ new TableSchema()
+ .setFields(
+ new ArrayList<>(
+ ImmutableList.of(
+ new TableFieldSchema().setName("description").setType("STRING")))));
+
+ p.apply(Create.of(inputs).withCoder(StringUtf8Coder.of()))
Review Comment:
Hm, since these are unit tests, and using different pipeline shapes, I think it would be better to keep them as separate tests (though there will be admittedly some overlap in the setup code.)
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Test sink name [beam]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #29230:
URL: https://github.com/apache/beam/pull/29230#issuecomment-1814326538
Reminder, please take a look at this pr: @chamikaramj
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org