[BEAM-2806] support View.CreatePCollectionView in FlinkRunner

• Type: New Feature
• Status: Resolved
• Priority: P2
• Resolution: Fixed
• Fix Version/s: 2.3.0
• Component/s: runner-flink
• Assignee: Grzegorz Kołakowski
• Reporter: Mingmin Xu
• Created: 24/Aug/17 23:30
• Updated: 16/May/20 13:56
• Resolved: 07/Feb/18 13:50

Description

Beam version: 2.2.0-SNAPSHOT

Here's the code

PCollectionView<Map<BeamRecord, Iterable<BeamRecord>>> rowsView = rightRows
        .apply(View.<BeamRecord, BeamRecord>asMultimap());

And exception when running with FlinkRunner:

Exception in thread "main" java.lang.UnsupportedOperationException: The transform View.CreatePCollectionView is currently not supported.
	at org.apache.beam.runners.flink.FlinkStreamingPipelineTranslator.visitPrimitiveTransform(FlinkStreamingPipelineTranslator.java:113)
	at org.apache.beam.sdk.runners.TransformHierarchy$Node.visit(TransformHierarchy.java:594)
	at org.apache.beam.sdk.runners.TransformHierarchy$Node.visit(TransformHierarchy.java:586)
	at org.apache.beam.sdk.runners.TransformHierarchy$Node.visit(TransformHierarchy.java:586)
	at org.apache.beam.sdk.runners.TransformHierarchy$Node.access$500(TransformHierarchy.java:268)
	at org.apache.beam.sdk.runners.TransformHierarchy.visit(TransformHierarchy.java:202)
	at org.apache.beam.sdk.Pipeline.traverseTopologically(Pipeline.java:440)
	at org.apache.beam.runners.flink.FlinkPipelineTranslator.translate(FlinkPipelineTranslator.java:38)
	at org.apache.beam.runners.flink.FlinkStreamingPipelineTranslator.translate(FlinkStreamingPipelineTranslator.java:69)
	at org.apache.beam.runners.flink.FlinkPipelineExecutionEnvironment.translate(FlinkPipelineExecutionEnvironment.java:104)
	at org.apache.beam.runners.flink.FlinkRunner.run(FlinkRunner.java:113)
	at org.apache.beam.sdk.Pipeline.run(Pipeline.java:297)
	at org.apache.beam.sdk.Pipeline.run(Pipeline.java:283)

Issue Links

• is related to: BEAM-3589 Flink runner breaks with ClassCastException on UnboundedSource
• links to: GitHub Pull Request #4558