[
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.
*
+ * <p>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.
+ *
* <p>The transform does not provide same transactional guarantees as Cloud
Spanner. In particular,
*
* <ul>
@@ -682,6 +687,9 @@ public CreateTransaction withTimestampBound(TimestampBound
timestampBound) {
abstract PTransform<PCollection<KV<String, byte[]>>,
PCollection<KV<String, List<byte[]>>>>
getSampler();
+ @Nullable
+ abstract PCollection getSchemaReadySignal();
+
abstract Builder toBuilder();
@AutoValue.Builder
@@ -701,6 +709,8 @@ abstract Builder setSampler(
PTransform<PCollection<KV<String, byte[]>>, PCollection<KV<String,
List<byte[]>>>>
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<Mutation> input) {
getSpannerConfig().validate();
@@ -835,13 +854,16 @@ public SpannerWriteResult
expand(PCollection<MutationGroup> input) {
if (sampler == null) {
sampler = createDefaultSampler();
}
+
// First, read the Cloud Spanner schema.
+ PCollection<Void> 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<SpannerSchema> 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:
[email protected]
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
> Priority: Major
> Fix For: 2.9.0
>
> Time Spent: 50m
> 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)
