You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by ar...@apache.org on 2019/02/04 10:02:22 UTC
[beam] branch spark-runner_structured-streaming updated: Don't use
deprecated sideInput.getWindowingStrategyInternal()
This is an automated email from the ASF dual-hosted git repository.
aromanenko pushed a commit to branch spark-runner_structured-streaming
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/spark-runner_structured-streaming by this push:
new e65815e Don't use deprecated sideInput.getWindowingStrategyInternal()
e65815e is described below
commit e65815e97b5056a839b0d9e2cce5543a7231d158
Author: Alexey Romanenko <ar...@gmail.com>
AuthorDate: Fri Feb 1 18:50:00 2019 +0100
Don't use deprecated sideInput.getWindowingStrategyInternal()
---
.../translation/batch/ParDoTranslatorBatch.java | 22 +++++++++++++---------
1 file changed, 13 insertions(+), 9 deletions(-)
diff --git a/runners/spark-structured-streaming/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/ParDoTranslatorBatch.java b/runners/spark-structured-streaming/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/ParDoTranslatorBatch.java
index a984615..fbb6649 100644
--- a/runners/spark-structured-streaming/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/ParDoTranslatorBatch.java
+++ b/runners/spark-structured-streaming/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/ParDoTranslatorBatch.java
@@ -61,6 +61,7 @@ class ParDoTranslatorBatch<InputT, OutputT>
public void translateTransform(
PTransform<PCollection<InputT>, PCollectionTuple> transform, TranslationContext context) {
+ // Check for not-supported advanced features
// TODO: add support of Splittable DoFn
DoFn<InputT, OutputT> doFn = getDoFn(context);
checkState(
@@ -74,10 +75,16 @@ class ParDoTranslatorBatch<InputT, OutputT>
signature.stateDeclarations().size() > 0 || signature.timerDeclarations().size() > 0;
checkState(!stateful, "States and timers are not supported for the moment.");
+ // TODO: add support of SideInputs
+ List<PCollectionView<?>> sideInputs = getSideInputs(context);
+ final boolean hasSideInputs = sideInputs != null && sideInputs.size() > 0;
+ checkState(!hasSideInputs, "SideInputs are not supported for the moment.");
+
+
+ // Init main variables
Dataset<WindowedValue<InputT>> inputDataSet = context.getDataset(context.getInput());
Map<TupleTag<?>, PValue> outputs = context.getOutputs();
TupleTag<?> mainOutputTag = getTupleTag(context);
-
Map<TupleTag<?>, Integer> outputTags = Maps.newHashMap();
outputTags.put(mainOutputTag, 0);
@@ -98,7 +105,7 @@ class ParDoTranslatorBatch<InputT, OutputT>
WindowingStrategy<?, ?> windowingStrategy = null;
// collect all output Coders and create a UnionCoder for our tagged outputs
- List<Coder<?>> outputCoders = Lists.newArrayList();
+// List<Coder<?>> outputCoders = Lists.newArrayList();
for (TupleTag<?> tag : indexMap.values()) {
PValue taggedValue = outputs.get(tag);
checkState(
@@ -107,7 +114,7 @@ class ParDoTranslatorBatch<InputT, OutputT>
taggedValue,
taggedValue.getClass().getSimpleName());
PCollection<?> coll = (PCollection<?>) taggedValue;
- outputCoders.add(coll.getCoder());
+// outputCoders.add(coll.getCoder());
windowingStrategy = coll.getWindowingStrategy();
}
@@ -115,18 +122,15 @@ class ParDoTranslatorBatch<InputT, OutputT>
throw new IllegalStateException("No outputs defined.");
}
- UnionCoder unionCoder = UnionCoder.of(outputCoders);
+// UnionCoder unionCoder = UnionCoder.of(outputCoders);
+
- List<PCollectionView<?>> sideInputs = getSideInputs(context);
- final boolean hasSideInputs = sideInputs != null && sideInputs.size() > 0;
- // TODO: add support of SideInputs
- checkState(!hasSideInputs, "SideInputs are not supported for the moment.");
// construct a map from side input to WindowingStrategy so that
// the DoFn runner can map main-input windows to side input windows
Map<PCollectionView<?>, WindowingStrategy<?, ?>> sideInputStrategies = new HashMap<>();
for (PCollectionView<?> sideInput : sideInputs) {
- sideInputStrategies.put(sideInput, sideInput.getWindowingStrategyInternal());
+ sideInputStrategies.put(sideInput, sideInput.getPCollection().getWindowingStrategy());
}
Map<TupleTag<?>, Coder<?>> outputCoderMap = context.getOutputCoders();