[jira] [Work logged] (BEAM-4796) SpannerIO waits for all input before writing
[ https://issues.apache.org/jira/browse/BEAM-4796?focusedWorklogId=153924&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-153924 ] ASF GitHub Bot logged work on BEAM-4796: Author: ASF GitHub Bot Created on: 12/Oct/18 16:03 Start Date: 12/Oct/18 16:03 Worklog Time Spent: 10m Work Description: nielm commented on issue #6409: [BEAM-4796] SpannerIO: Add option to wait for Schema to be ready. URL: https://github.com/apache/beam/pull/6409#issuecomment-429376495 > Can we close this PR given that #6478 includes this ? yes, closed. This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org Issue Time Tracking --- Worklog Id: (was: 153924) Time Spent: 40m (was: 0.5h) > SpannerIO waits for all input before writing > > > Key: BEAM-4796 > URL: https://issues.apache.org/jira/browse/BEAM-4796 > Project: Beam > Issue Type: Bug > Components: io-java-gcp >Affects Versions: 2.5.0 >Reporter: Niel Markwick >Assignee: Niel Markwick >Priority: Major > Fix For: 2.9.0 > > Time Spent: 40m > Remaining Estimate: 0h > > SpannerIO.Write waits for all input in the window to arrive before getting > the schema: > [https://github.com/apache/beam/blame/release-2.5.0/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/SpannerIO.java#L841] > > In streaming mode, this is not an issue, but in batch mode, this causes the > pipeline to stall until all input is read, which could be a significant > amount of time (and temp data). > > -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Work logged] (BEAM-4796) SpannerIO waits for all input before writing
[
https://issues.apache.org/jira/browse/BEAM-4796?focusedWorklogId=153925&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-153925
]
ASF GitHub Bot logged work on BEAM-4796:
Author: ASF GitHub Bot
Created on: 12/Oct/18 16:03
Start Date: 12/Oct/18 16:03
Worklog Time Spent: 10m
Work Description: nielm closed pull request #6409: [BEAM-4796] SpannerIO:
Add option to wait for Schema to be ready.
URL: https://github.com/apache/beam/pull/6409
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git
a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/SpannerIO.java
b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/SpannerIO.java
index af2a3b00d1f..5cb2469bcfb 100644
---
a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/SpannerIO.java
+++
b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/SpannerIO.java
@@ -163,6 +163,11 @@
* Write#withBatchSizeBytes(long)}. Setting batch size to a small value or
zero practically disables
* batching.
*
+ * The write transform reads the database schema on pipeline start. If the
schema is created as
+ * part of the same pipline, this transform needs to wait until the schema has
been created. Use
+ * {@link Write#withSchemaReadySignal(PCollection)} to pass a {@link
PCollection} which will be used
+ * with {@link Wait#on(PCollection[])} to prevent the schema from being read
until it is ready.
+ *
* The transform does not provide same transactional guarantees as Cloud
Spanner. In particular,
*
*
@@ -682,6 +687,9 @@ public CreateTransaction withTimestampBound(TimestampBound
timestampBound) {
abstract PTransform>,
PCollection>>>
getSampler();
+@Nullable
+abstract PCollection getSchemaReadySignal();
+
abstract Builder toBuilder();
@AutoValue.Builder
@@ -701,6 +709,8 @@ abstract Builder setSampler(
PTransform>, PCollection>>>
sampler);
+ abstract Builder setSchemaReadySignal(PCollection schemaReadySignal);
+
abstract Write build();
}
@@ -786,6 +796,15 @@ public Write withMaxNumMutations(long maxNumMutations) {
return toBuilder().setMaxNumMutations(maxNumMutations).build();
}
+/**
+ * Specifies an input PCollection that can be used with a {@code
Wait.on(signal)} to indicate
+ * when the database schema is ready. To be used when the schema creation
is part of the
+ * pipeline to prevent the connector reading the schema too early.
+ */
+public Write withSchemaReadySignal(PCollection signal) {
+ return toBuilder().setSchemaReadySignal(signal).build();
+}
+
@Override
public SpannerWriteResult expand(PCollection input) {
getSpannerConfig().validate();
@@ -835,13 +854,16 @@ public SpannerWriteResult
expand(PCollection input) {
if (sampler == null) {
sampler = createDefaultSampler();
}
+
// First, read the Cloud Spanner schema.
+ PCollection schemaSeed =
+ input.getPipeline().apply("Create Seed", Create.of((Void) null));
+ if (spec.getSchemaReadySignal() != null) {
+// Wait for external signal before reading schema.
+schemaSeed = schemaSeed.apply("Wait for schema",
Wait.on(spec.getSchemaReadySignal()));
+ }
final PCollectionView schemaView =
- input
- .getPipeline()
- .apply("Create seed", Create.of((Void) null))
- // Wait for input mutations so it is possible to chain
transforms.
- .apply(Wait.on(input))
+ schemaSeed
.apply(
"Read information schema",
ParDo.of(new ReadSpannerSchema(spec.getSpannerConfig(
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
Issue Time Tracking
---
Worklog Id: (was: 153925)
Time Spent: 50m (was: 40m)
> SpannerIO waits for all input before writing
>
>
> Key: BEAM-4796
> URL: https://issues.apache.org/jira/browse/BEAM-4796
> Project: Beam
> Issue Type: Bug
> Components: io-java-gcp
>Affects Versions: 2.5.0
>Reporter: Niel Markwick
>Assignee: Niel Markwick
>Prio
[jira] [Work logged] (BEAM-4796) SpannerIO waits for all input before writing
[ https://issues.apache.org/jira/browse/BEAM-4796?focusedWorklogId=153228&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-153228 ] ASF GitHub Bot logged work on BEAM-4796: Author: ASF GitHub Bot Created on: 10/Oct/18 18:05 Start Date: 10/Oct/18 18:05 Worklog Time Spent: 10m Work Description: chamikaramj commented on issue #6409: [BEAM-4796] SpannerIO: Add option to wait for Schema to be ready. URL: https://github.com/apache/beam/pull/6409#issuecomment-428672741 Can we close this PR given that https://github.com/apache/beam/pull/6478 includes this ? This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org Issue Time Tracking --- Worklog Id: (was: 153228) Time Spent: 0.5h (was: 20m) > SpannerIO waits for all input before writing > > > Key: BEAM-4796 > URL: https://issues.apache.org/jira/browse/BEAM-4796 > Project: Beam > Issue Type: Bug > Components: io-java-gcp >Affects Versions: 2.5.0 >Reporter: Niel Markwick >Assignee: Niel Markwick >Priority: Major > Fix For: 2.9.0 > > Time Spent: 0.5h > Remaining Estimate: 0h > > SpannerIO.Write waits for all input in the window to arrive before getting > the schema: > [https://github.com/apache/beam/blame/release-2.5.0/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/SpannerIO.java#L841] > > In streaming mode, this is not an issue, but in batch mode, this causes the > pipeline to stall until all input is read, which could be a significant > amount of time (and temp data). > > -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Work logged] (BEAM-4796) SpannerIO waits for all input before writing
[ https://issues.apache.org/jira/browse/BEAM-4796?focusedWorklogId=144875&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-144875 ] ASF GitHub Bot logged work on BEAM-4796: Author: ASF GitHub Bot Created on: 17/Sep/18 15:25 Start Date: 17/Sep/18 15:25 Worklog Time Spent: 10m Work Description: nielm commented on issue #6409: [BEAM-4796] SpannerIO: Add option to wait for Schema to be ready. URL: https://github.com/apache/beam/pull/6409#issuecomment-422061192 Note that I could not work out how to test a pipeline that uses Wait.on(signal), so no tests were added for this PR. This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org Issue Time Tracking --- Worklog Id: (was: 144875) Time Spent: 20m (was: 10m) > SpannerIO waits for all input before writing > > > Key: BEAM-4796 > URL: https://issues.apache.org/jira/browse/BEAM-4796 > Project: Beam > Issue Type: Bug > Components: io-java-gcp >Affects Versions: 2.5.0 >Reporter: Niel Markwick >Assignee: Chamikara Jayalath >Priority: Major > Time Spent: 20m > Remaining Estimate: 0h > > SpannerIO.Write waits for all input in the window to arrive before getting > the schema: > [https://github.com/apache/beam/blame/release-2.5.0/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/SpannerIO.java#L841] > > In streaming mode, this is not an issue, but in batch mode, this causes the > pipeline to stall until all input is read, which could be a significant > amount of time (and temp data). > > -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Work logged] (BEAM-4796) SpannerIO waits for all input before writing
[ https://issues.apache.org/jira/browse/BEAM-4796?focusedWorklogId=144873&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-144873 ] ASF GitHub Bot logged work on BEAM-4796: Author: ASF GitHub Bot Created on: 17/Sep/18 15:22 Start Date: 17/Sep/18 15:22 Worklog Time Spent: 10m Work Description: nielm opened a new pull request #6409: [BEAM-4796] SpannerIO: Add option to wait for Schema to be ready. URL: https://github.com/apache/beam/pull/6409 Current behavior waits for the entire input PCollection to be read and closed before reading the Schema. This can delay the pipeline for large inputs, and does not guarantee that the schema is ready (if it is created in the same pipeline) for small inputs. It also breaks streaming mode completely as the input PCollection is never closed. This PR adds an optional parameter with a PCollection to wait for before reading the schema. If not specified, the schema is read immediately. This provides a partial -- but not complete -- fix for streaming mode (there are still issues with the partitioning/grouping in streaming mode which means that NPE's will be thrown with more than trivial load). @chamikaramj Post-Commit Tests Status (on master branch) Lang | SDK | Apex | Dataflow | Flink | Gearpump | Samza | Spark --- | --- | --- | --- | --- | --- | --- | --- Go | [](https://builds.apache.org/job/beam_PostCommit_Go_GradleBuild/lastCompletedBuild/) | --- | --- | --- | --- | --- | --- Java | [](https://builds.apache.org/job/beam_PostCommit_Java_GradleBuild/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Apex_Gradle/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow_Gradle/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Flink_Gradle/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Gearpump_Gradle/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Samza_Gradle/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Spark_Gradle/lastCompletedBuild/) Python | [](https://builds.apache.org/job/beam_PostCommit_Python_Verify/lastCompletedBuild/) | --- | [](https://builds.apache.org/job/beam_PostCommit_Py_VR_Dataflow/lastCompletedBuild/) [](https://builds.apache.org/job/beam_PostCommit_Py_ValCont/lastCompletedBuild/) | --- | --- | --- | --- This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org Issue Time Tracking --- Worklog Id: (was: 144873) Time Spent: 10m Remaining Estimate: 0h > SpannerIO waits for all input before writing > > > Key: BEAM-4796 > URL: https://issues.apache.org/jira/browse/BEAM-4796 > Project: Beam > Issue Type: Bug > Components: io-java-gcp >Affects Versions: 2.5.0 >Reporter: Niel Markwick >Assignee: Chamikara Jayalath >Priority: Major >
