beam-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From "ASF GitHub Bot (JIRA)" <j...@apache.org>
Subject [jira] [Work logged] (BEAM-4297) Flink portable runner executable stage operator for streaming
Date Wed, 23 May 2018 18:09:00 GMT

     [ https://issues.apache.org/jira/browse/BEAM-4297?focusedWorklogId=105209&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-105209
]

ASF GitHub Bot logged work on BEAM-4297:
----------------------------------------

                Author: ASF GitHub Bot
            Created on: 23/May/18 18:08
            Start Date: 23/May/18 18:08
    Worklog Time Spent: 10m 
      Work Description: lukecwik commented on a change in pull request #5407: [BEAM-4297]
Streaming executable stage translation and operator for portable Flink runner.
URL: https://github.com/apache/beam/pull/5407#discussion_r190338072
 
 

 ##########
 File path: runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkStreamingPortablePipelineTranslator.java
 ##########
 @@ -423,8 +432,133 @@ private void translateImpulse(
           String id,
           RunnerApi.Pipeline pipeline,
           StreamingTranslationContext context) {
+    // TODO: Fail on stateful DoFns for now.
+    // TODO: Support stateful DoFns by inserting group-by-keys where necessary.
+    // TODO: Fail on splittable DoFns.
+    // TODO: Special-case single outputs to avoid multiplexing PCollections.
+    RunnerApi.Components components = pipeline.getComponents();
+    RunnerApi.PTransform transform = components.getTransformsOrThrow(id);
+    Map<String, String> outputs = transform.getOutputsMap();
+    RehydratedComponents rehydratedComponents =
+            RehydratedComponents.forComponents(components);
+
+    BiMap<String, Integer> outputMap =
+            FlinkPipelineTranslatorUtils.createOutputMap(outputs.keySet());
+    Map<String, Coder<WindowedValue<?>>> outputCoders = Maps.newHashMap();
+    for (String localOutputName : new TreeMap<>(outputMap.inverse()).values()) {
+      String collectionId = outputs.get(localOutputName);
+      Coder<WindowedValue<?>> windowCoder = (Coder) instantiateCoder(collectionId,
components);
+      outputCoders.put(localOutputName, windowCoder);
+    }
+
+    final RunnerApi.ExecutableStagePayload stagePayload;
+    try {
+      stagePayload = RunnerApi.ExecutableStagePayload.parseFrom(transform.getSpec().getPayload());
+    } catch (IOException e) {
+      throw new RuntimeException(e);
+    }
+
+    String inputPCollectionId =
+            Iterables.getOnlyElement(transform.getInputsMap().values());
 
-    throw new RuntimeException("executable stage translation not implemented");
+    // TODO: is this still relevant?
+    // we assume that the transformation does not change the windowing strategy.
+    RunnerApi.WindowingStrategy windowingStrategyProto =
 
 Review comment:
   ExecutableStage's can change the windowing strategy as they may execute assign windows
within. So a previous executable stage may have changed the windowing strategy.

----------------------------------------------------------------
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:
users@infra.apache.org


Issue Time Tracking
-------------------

    Worklog Id:     (was: 105209)

> Flink portable runner executable stage operator for streaming
> -------------------------------------------------------------
>
>                 Key: BEAM-4297
>                 URL: https://issues.apache.org/jira/browse/BEAM-4297
>             Project: Beam
>          Issue Type: Task
>          Components: runner-flink
>            Reporter: Thomas Weise
>            Assignee: Thomas Weise
>            Priority: Major
>              Labels: portability
>          Time Spent: 1h
>  Remaining Estimate: 0h
>




--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

Mime
View raw message