beam-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From "Xu Mingmin (JIRA)" <j...@apache.org>
Subject [jira] [Commented] (BEAM-2806) support View.CreatePCollectionView in FlinkRunner
Date Fri, 25 Aug 2017 21:08:00 GMT

    [ https://issues.apache.org/jira/browse/BEAM-2806?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16142200#comment-16142200
] 

Xu Mingmin commented on BEAM-2806:
----------------------------------

Sure, this is the code we use in SQL join
{code}
    PipelineOptions options = PipelineOptionsFactory.fromArgs(new String[]{}).as(PipelineOptions.class);
    options.setRunner(FlinkRunner.class);
    
    Pipeline p = Pipeline.create(options);
    PCollection<BeamRecord> streamingPCollection = sojEventTable.buildIOReader(p);
    PCollection<BeamRecord> lkpPCollection = lkpTable.buildIOReader(p);
    
    final PCollectionView<Map<Integer, Iterable<BeamRecord>>> lkpAsView
= lkpPCollection
        .apply(WithKeys.<Integer, BeamRecord>of(new SerializableFunction<BeamRecord,
Integer>() {
          @Override
          public Integer apply(BeamRecord input) {
            return input.getInteger(0);
          }
        }))
        .apply(View.<Integer, BeamRecord>asMultimap());
    
    PCollection<Void> ret = streamingPCollection.apply(
        ParDo.of(new DoFn<BeamRecord, Void>(){
          @ProcessElement public void processElement(ProcessContext context) {
            BeamRecord drvRow = context.element();
            Map<Integer, Iterable<BeamRecord>> key2Rows = context.sideInput(lkpAsView);
            int pageId = drvRow.getInteger("PAGE_ID");
            if(key2Rows.get(pageId) != null){
              System.out.println("Record Pass: "+drvRow);
            }
          }
        }).withSideInputs(lkpAsView)
        );

    p.run().waitUntilFinish();
{code}


> support View.CreatePCollectionView in FlinkRunner
> -------------------------------------------------
>
>                 Key: BEAM-2806
>                 URL: https://issues.apache.org/jira/browse/BEAM-2806
>             Project: Beam
>          Issue Type: New Feature
>          Components: runner-flink
>            Reporter: Xu Mingmin
>            Assignee: Aljoscha Krettek
>
> Beam version: 2.2.0-SNAPSHOT
> Here's the code
> {code}
> PCollectionView<Map<BeamRecord, Iterable<BeamRecord>>> rowsView = rightRows
>         .apply(View.<BeamRecord, BeamRecord>asMultimap());
> {code}
> And exception when running with {{FlinkRunner}}:
> {code}
> 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)
> {code}



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

Mime
View raw message