You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by GitBox <gi...@apache.org> on 2021/08/25 14:13:58 UTC
[GitHub] [beam] je-ik commented on a change in pull request #15370: [BEAM-12704] flink primitive read
je-ik commented on a change in pull request #15370:
URL: https://github.com/apache/beam/pull/15370#discussion_r695791725
##########
File path: runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/External.java
##########
@@ -340,8 +340,7 @@ public OutputT expand(InputT input) {
boolean isJavaSDKCompatible(RunnerApi.Components components, String coderId) {
RunnerApi.Coder coder = components.getCodersOrThrow(coderId);
- if (!CoderTranslation.JAVA_SERIALIZED_CODER_URN.equals(coder.getSpec().getUrn())
- && !CoderTranslation.KNOWN_CODER_URNS.containsValue(coder.getSpec().getUrn())) {
+ if (!CoderTranslation.isKnownCoderURN(coder.getSpec().getUrn())) {
Review comment:
This was changed when I was trying to get `beam:coders:java:0.1` URN to `CoderTranslation`. It is not needed, reverted.
##########
File path: runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkStreamingPortablePipelineTranslator.java
##########
@@ -534,19 +594,55 @@ public void flatMap(T t, Collector<T> collector) {
source =
nonDedupSource
.keyBy(new FlinkStreamingTransformTranslators.ValueWithRecordIdKeySelector<>())
- .transform("deduping", outputTypeInfo, new DedupingOperator<>(pipelineOptions))
- .uid(format("%s/__deduplicated__", transformName));
+ .transform("deduping", sdkTypeInformation, new DedupingOperator<>(pipelineOptions))
+ .uid(format("%s/__deduplicated__", transformName))
+ .returns(sdkTypeInformation);
} else {
source =
nonDedupSource
.flatMap(new FlinkStreamingTransformTranslators.StripIdsMap<>(pipelineOptions))
- .returns(outputTypeInfo);
+ .returns(sdkTypeInformation);
}
+
+ return source.map(value -> intoWireTypes(sdkCoder, wireCoder, value)).returns(outputTypeInfo);
} catch (Exception e) {
throw new RuntimeException("Error while translating UnboundedSource: " + unboundedSource, e);
}
+ }
- return source;
+ private static <T> WindowedValue.FullWindowedValueCoder<T> getSdkCoder(
Review comment:
done
##########
File path: runners/flink/src/main/java/org/apache/beam/runners/flink/translation/wrappers/SourceInputFormat.java
##########
@@ -150,7 +151,8 @@ public boolean reachedEnd() throws IOException {
@Override
public void close() throws IOException {
- metricContainer.registerMetricsForPipelineResult();
+ Optional.ofNullable(metricContainer)
+ .ifPresent(FlinkMetricContainer::registerMetricsForPipelineResult);
Review comment:
fixed
##########
File path: runners/flink/src/test/java/org/apache/beam/runners/flink/ReadSourcePortableTest.java
##########
@@ -81,7 +96,8 @@ public static void tearDown() throws InterruptedException {
@Test(timeout = 120_000)
public void testExecution() throws Exception {
- PipelineOptions options = PipelineOptionsFactory.fromArgs("--experiments=beam_fn_api").create();
+ PipelineOptions options =
+ PipelineOptionsFactory.fromArgs("--experiments=use_deprecated_read").create();
Review comment:
I think that is tested in other tests.
##########
File path: runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkStreamingPortablePipelineTranslator.java
##########
@@ -534,19 +594,55 @@ public void flatMap(T t, Collector<T> collector) {
source =
nonDedupSource
.keyBy(new FlinkStreamingTransformTranslators.ValueWithRecordIdKeySelector<>())
- .transform("deduping", outputTypeInfo, new DedupingOperator<>(pipelineOptions))
- .uid(format("%s/__deduplicated__", transformName));
+ .transform("deduping", sdkTypeInformation, new DedupingOperator<>(pipelineOptions))
+ .uid(format("%s/__deduplicated__", transformName))
+ .returns(sdkTypeInformation);
} else {
source =
nonDedupSource
.flatMap(new FlinkStreamingTransformTranslators.StripIdsMap<>(pipelineOptions))
- .returns(outputTypeInfo);
+ .returns(sdkTypeInformation);
}
+
+ return source.map(value -> intoWireTypes(sdkCoder, wireCoder, value)).returns(outputTypeInfo);
} catch (Exception e) {
throw new RuntimeException("Error while translating UnboundedSource: " + unboundedSource, e);
}
+ }
- return source;
+ private static <T> WindowedValue.FullWindowedValueCoder<T> getSdkCoder(
+ String pCollectionId, RunnerApi.Components components) {
+
+ PipelineNode.PCollectionNode pCollectionNode =
+ PipelineNode.pCollection(pCollectionId, components.getPcollectionsOrThrow(pCollectionId));
+ RunnerApi.Components.Builder componentsBuilder = components.toBuilder();
+ String coderId =
+ WireCoders.addSdkWireCoder(
+ pCollectionNode,
+ componentsBuilder,
+ RunnerApi.ExecutableStagePayload.WireCoderSetting.getDefaultInstance());
+ RehydratedComponents rehydratedComponents =
+ RehydratedComponents.forComponents(componentsBuilder.build());
+ try {
+ @SuppressWarnings("unchecked")
+ WindowedValue.FullWindowedValueCoder<T> res =
+ (WindowedValue.FullWindowedValueCoder<T>) rehydratedComponents.getCoder(coderId);
+ return res;
+ } catch (IOException ex) {
+ throw new IllegalStateException(ex);
+ }
+ }
+
+ private static <InputT, OutputT> WindowedValue<OutputT> intoWireTypes(
+ Coder<WindowedValue<InputT>> inCoder,
+ Coder<WindowedValue<OutputT>> outCoder,
+ WindowedValue<InputT> value) {
+
+ try {
+ return CoderUtils.decodeFromByteArray(outCoder, CoderUtils.encodeToByteArray(inCoder, value));
+ } catch (CoderException ex) {
+ throw new IllegalStateException(ex);
Review comment:
I don't see any added value of the message, the cause will be visible from the stack trace.
##########
File path: runners/flink/src/main/java/org/apache/beam/runners/flink/translation/types/CoderTypeSerializer.java
##########
@@ -86,12 +86,11 @@ public T createInstance() {
public T copy(T t) {
if (fasterCopy) {
return t;
- } else {
- try {
- return CoderUtils.clone(coder, t);
- } catch (CoderException e) {
- throw new RuntimeException("Could not clone.", e);
- }
+ }
+ try {
+ return CoderUtils.clone(coder, t);
+ } catch (Exception e) {
+ throw new RuntimeException("Could not clone value " + t + " using " + coder, e);
Review comment:
part of debugging, reverted.
##########
File path: runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkStreamingPortablePipelineTranslator.java
##########
@@ -534,19 +594,55 @@ public void flatMap(T t, Collector<T> collector) {
source =
nonDedupSource
.keyBy(new FlinkStreamingTransformTranslators.ValueWithRecordIdKeySelector<>())
- .transform("deduping", outputTypeInfo, new DedupingOperator<>(pipelineOptions))
- .uid(format("%s/__deduplicated__", transformName));
+ .transform("deduping", sdkTypeInformation, new DedupingOperator<>(pipelineOptions))
+ .uid(format("%s/__deduplicated__", transformName))
+ .returns(sdkTypeInformation);
} else {
source =
nonDedupSource
.flatMap(new FlinkStreamingTransformTranslators.StripIdsMap<>(pipelineOptions))
- .returns(outputTypeInfo);
+ .returns(sdkTypeInformation);
}
+
+ return source.map(value -> intoWireTypes(sdkCoder, wireCoder, value)).returns(outputTypeInfo);
} catch (Exception e) {
throw new RuntimeException("Error while translating UnboundedSource: " + unboundedSource, e);
}
+ }
- return source;
+ private static <T> WindowedValue.FullWindowedValueCoder<T> getSdkCoder(
+ String pCollectionId, RunnerApi.Components components) {
+
+ PipelineNode.PCollectionNode pCollectionNode =
+ PipelineNode.pCollection(pCollectionId, components.getPcollectionsOrThrow(pCollectionId));
+ RunnerApi.Components.Builder componentsBuilder = components.toBuilder();
+ String coderId =
+ WireCoders.addSdkWireCoder(
+ pCollectionNode,
+ componentsBuilder,
+ RunnerApi.ExecutableStagePayload.WireCoderSetting.getDefaultInstance());
+ RehydratedComponents rehydratedComponents =
+ RehydratedComponents.forComponents(componentsBuilder.build());
+ try {
+ @SuppressWarnings("unchecked")
+ WindowedValue.FullWindowedValueCoder<T> res =
+ (WindowedValue.FullWindowedValueCoder<T>) rehydratedComponents.getCoder(coderId);
+ return res;
+ } catch (IOException ex) {
+ throw new IllegalStateException(ex);
+ }
+ }
+
+ private static <InputT, OutputT> WindowedValue<OutputT> intoWireTypes(
+ Coder<WindowedValue<InputT>> inCoder,
+ Coder<WindowedValue<OutputT>> outCoder,
+ WindowedValue<InputT> value) {
+
+ try {
+ return CoderUtils.decodeFromByteArray(outCoder, CoderUtils.encodeToByteArray(inCoder, value));
Review comment:
It definitely is suboptimal, but given the performance penalty of portability itself I seriously doubt this will be addressed separately. This can be optimized when (and if) we get to the stage that we will unify classical and portable runners (so that classical runners will be portable runners that are able to inline complete pipeline, because the pipeline is written in the same SDK as the runner).
##########
File path: runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkStreamingPortablePipelineTranslator.java
##########
@@ -500,28 +547,41 @@ public void flatMap(T t, Collector<T> collector) {
final DataStream<WindowedValue<T>> source;
final DataStream<WindowedValue<ValueWithRecordId<T>>> nonDedupSource;
- Coder<WindowedValue<T>> windowCoder =
- instantiateCoder(outputCollectionId, pipeline.getComponents());
- TypeInformation<WindowedValue<T>> outputTypeInfo =
- new CoderTypeInformation<>(windowCoder, pipelineOptions);
+ @SuppressWarnings("unchecked")
+ UnboundedSource<T, ?> unboundedSource =
+ (UnboundedSource<T, ?>) ReadTranslation.unboundedSourceFromProto(payload);
- WindowingStrategy windowStrategy =
+ @SuppressWarnings("unchecked")
+ WindowingStrategy<T, ?> windowStrategy =
getWindowingStrategy(outputCollectionId, pipeline.getComponents());
- TypeInformation<WindowedValue<ValueWithRecordId<T>>> withIdTypeInfo =
- new CoderTypeInformation<>(
- WindowedValue.getFullCoder(
- ValueWithRecordId.ValueWithRecordIdCoder.of(
- ((WindowedValueCoder) windowCoder).getValueCoder()),
- windowStrategy.getWindowFn().windowCoder()),
- pipelineOptions);
-
- UnboundedSource unboundedSource = ReadTranslation.unboundedSourceFromProto(payload);
try {
+
+ @SuppressWarnings("unchecked")
+ WindowedValue.FullWindowedValueCoder<T> wireCoder =
+ (WindowedValue.FullWindowedValueCoder<T>)
+ (Coder) instantiateCoder(outputCollectionId, pipeline.getComponents());
Review comment:
fixed
##########
File path: runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkStreamingPortablePipelineTranslator.java
##########
@@ -474,20 +477,64 @@ public void flatMap(T t, Collector<T> collector) {
throw new RuntimeException("Failed to parse ReadPayload from transform", e);
}
- Preconditions.checkState(
- payload.getIsBounded() != RunnerApi.IsBounded.Enum.BOUNDED,
- "Bounded reads should run inside an environment instead of being translated by the Runner.");
+ final DataStream<WindowedValue<T>> source;
+ if (payload.getIsBounded() == RunnerApi.IsBounded.Enum.BOUNDED) {
+ source =
+ translateBoundedSource(
+ transform.getUniqueName(),
+ outputCollectionId,
+ payload,
+ pipeline,
+ context.getPipelineOptions(),
+ context.getExecutionEnvironment());
+ } else {
+ source =
+ translateUnboundedSource(
+ transform.getUniqueName(),
+ outputCollectionId,
+ payload,
+ pipeline,
+ context.getPipelineOptions(),
+ context.getExecutionEnvironment());
+ }
+ context.addDataStream(outputCollectionId, source);
+ }
- DataStream<WindowedValue<T>> source =
- translateUnboundedSource(
- transform.getUniqueName(),
- outputCollectionId,
- payload,
- pipeline,
- context.getPipelineOptions(),
- context.getExecutionEnvironment());
+ private <T> DataStream<WindowedValue<T>> translateBoundedSource(
+ String transformName,
+ String outputCollectionId,
+ RunnerApi.ReadPayload payload,
+ RunnerApi.Pipeline pipeline,
+ FlinkPipelineOptions pipelineOptions,
+ StreamExecutionEnvironment env) {
- context.addDataStream(outputCollectionId, source);
+ try {
+ @SuppressWarnings("unchecked")
+ BoundedSource<T> boundedSource =
+ (BoundedSource<T>) ReadTranslation.boundedSourceFromProto(payload);
+ @SuppressWarnings("unchecked")
+ WindowedValue.FullWindowedValueCoder<T> wireCoder =
+ (WindowedValue.FullWindowedValueCoder<T>)
Review comment:
fixed
##########
File path: runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkStreamingPortablePipelineTranslator.java
##########
@@ -534,19 +594,55 @@ public void flatMap(T t, Collector<T> collector) {
source =
nonDedupSource
.keyBy(new FlinkStreamingTransformTranslators.ValueWithRecordIdKeySelector<>())
- .transform("deduping", outputTypeInfo, new DedupingOperator<>(pipelineOptions))
- .uid(format("%s/__deduplicated__", transformName));
+ .transform("deduping", sdkTypeInformation, new DedupingOperator<>(pipelineOptions))
+ .uid(format("%s/__deduplicated__", transformName))
+ .returns(sdkTypeInformation);
} else {
source =
nonDedupSource
.flatMap(new FlinkStreamingTransformTranslators.StripIdsMap<>(pipelineOptions))
- .returns(outputTypeInfo);
+ .returns(sdkTypeInformation);
}
+
+ return source.map(value -> intoWireTypes(sdkCoder, wireCoder, value)).returns(outputTypeInfo);
} catch (Exception e) {
throw new RuntimeException("Error while translating UnboundedSource: " + unboundedSource, e);
}
+ }
- return source;
+ private static <T> WindowedValue.FullWindowedValueCoder<T> getSdkCoder(
+ String pCollectionId, RunnerApi.Components components) {
+
+ PipelineNode.PCollectionNode pCollectionNode =
+ PipelineNode.pCollection(pCollectionId, components.getPcollectionsOrThrow(pCollectionId));
+ RunnerApi.Components.Builder componentsBuilder = components.toBuilder();
+ String coderId =
+ WireCoders.addSdkWireCoder(
+ pCollectionNode,
+ componentsBuilder,
+ RunnerApi.ExecutableStagePayload.WireCoderSetting.getDefaultInstance());
+ RehydratedComponents rehydratedComponents =
+ RehydratedComponents.forComponents(componentsBuilder.build());
+ try {
+ @SuppressWarnings("unchecked")
+ WindowedValue.FullWindowedValueCoder<T> res =
+ (WindowedValue.FullWindowedValueCoder<T>) rehydratedComponents.getCoder(coderId);
+ return res;
+ } catch (IOException ex) {
+ throw new IllegalStateException(ex);
Review comment:
same as below, the message will not bring any more information here.
##########
File path: runners/flink/src/test/java/org/apache/beam/runners/flink/ReadSourcePortableTest.java
##########
@@ -90,10 +106,16 @@ public void testExecution() throws Exception {
.as(PortablePipelineOptions.class)
.setDefaultEnvironmentType(Environments.ENVIRONMENT_EMBEDDED);
Pipeline p = Pipeline.create(options);
- PCollection<Long> result = p.apply(GenerateSequence.from(0L).to(10L));
+ PCollection<Long> result =
+ p.apply(Read.from(new Source(10)))
+ // FIXME: the test fails without this
Review comment:
That is unrelated, there is some bug in TrivialNativeTransformExpander or GreedyPipelineFuser, or somewhere in between.
##########
File path: runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkStreamingPortablePipelineTranslator.java
##########
@@ -500,28 +547,41 @@ public void flatMap(T t, Collector<T> collector) {
final DataStream<WindowedValue<T>> source;
final DataStream<WindowedValue<ValueWithRecordId<T>>> nonDedupSource;
- Coder<WindowedValue<T>> windowCoder =
- instantiateCoder(outputCollectionId, pipeline.getComponents());
- TypeInformation<WindowedValue<T>> outputTypeInfo =
- new CoderTypeInformation<>(windowCoder, pipelineOptions);
+ @SuppressWarnings("unchecked")
+ UnboundedSource<T, ?> unboundedSource =
+ (UnboundedSource<T, ?>) ReadTranslation.unboundedSourceFromProto(payload);
- WindowingStrategy windowStrategy =
+ @SuppressWarnings("unchecked")
+ WindowingStrategy<T, ?> windowStrategy =
getWindowingStrategy(outputCollectionId, pipeline.getComponents());
- TypeInformation<WindowedValue<ValueWithRecordId<T>>> withIdTypeInfo =
- new CoderTypeInformation<>(
- WindowedValue.getFullCoder(
- ValueWithRecordId.ValueWithRecordIdCoder.of(
- ((WindowedValueCoder) windowCoder).getValueCoder()),
- windowStrategy.getWindowFn().windowCoder()),
- pipelineOptions);
-
- UnboundedSource unboundedSource = ReadTranslation.unboundedSourceFromProto(payload);
try {
+
+ @SuppressWarnings("unchecked")
+ WindowedValue.FullWindowedValueCoder<T> wireCoder =
+ (WindowedValue.FullWindowedValueCoder<T>)
+ (Coder) instantiateCoder(outputCollectionId, pipeline.getComponents());
+
+ WindowedValue.FullWindowedValueCoder<T> sdkCoder =
+ getSdkCoder(outputCollectionId, pipeline.getComponents());
+
+ CoderTypeInformation<WindowedValue<T>> outputTypeInfo =
+ new CoderTypeInformation<>(wireCoder, pipelineOptions);
+
+ CoderTypeInformation<WindowedValue<T>> sdkTypeInformation =
+ new CoderTypeInformation<>(sdkCoder, pipelineOptions);
+
+ TypeInformation<WindowedValue<ValueWithRecordId<T>>> withIdTypeInfo =
+ new CoderTypeInformation<>(
+ WindowedValue.getFullCoder(
+ ValueWithRecordId.ValueWithRecordIdCoder.of(sdkCoder.getValueCoder()),
+ windowStrategy.getWindowFn().windowCoder()),
+ pipelineOptions);
+
int parallelism =
Review comment:
that is not related to this change
--
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