[BEAM-4297] Streaming executable stage translation and operator for portable Flink runner. by tweise · Pull Request #5407 · apache/beam

tweise · 2018-05-18T02:37:45Z

Executable stage translation for streaming mode based on the generic Flink streaming operator. Stage execution and tests adopted from batch translation.

Follow this checklist to help us incorporate your contribution quickly and easily:

Format the pull request title like [BEAM-XXX] Fixes bug in ApproximateQuantiles, where you replace BEAM-XXX with the appropriate JIRA issue, if applicable. This will automatically link the pull request to the issue.
If this contribution is large, please file an Apache Individual Contributor License Agreement.

It will help us expedite review of your Pull Request if you tag someone (e.g. @username) to look at it.

tweise · 2018-05-18T15:47:21Z

R: @lukecwik @bsidhom

bsidhom · 2018-05-18T16:52:18Z

.../flink/src/main/java/org/apache/beam/runners/flink/FlinkBatchPortablePipelineTranslator.java


  /**  Creates a mapping from PCollection id to output tag integer. */
-  private static BiMap<String, Integer> createOutputMap(Iterable<String> localOutputs) {
+  static BiMap<String, Integer> createOutputMap(Iterable<String> localOutputs) {


Please move this into a shared utility class.

tweise · 2018-05-23T15:18:39Z

@bsidhom @lukecwik PTAL

lukecwik · 2018-05-23T17:33:28Z

...nk/src/main/java/org/apache/beam/runners/flink/FlinkStreamingPortablePipelineTranslator.java

+    BiMap<String, Integer> outputMap =
+            FlinkPipelineTranslatorUtils.createOutputMap(outputs.keySet());
+    Map<String, Coder<WindowedValue<?>>> outputCoders = Maps.newHashMap();
+    for (String localOutputName : new TreeMap<>(outputMap.inverse()).values()) {


consider using TreeSet

A tree set of what? The intent here is to order the coders (the values in the map) by their tags (the keys).

I see, that makes sense.

I have cleaned up this portion of the code.

lukecwik · 2018-05-23T17:38:40Z

...nk/src/main/java/org/apache/beam/runners/flink/FlinkStreamingPortablePipelineTranslator.java

+    }
+
+    String inputPCollectionId =
+            Iterables.getOnlyElement(transform.getInputsMap().values());


Use ExecutableStage and rely on using getInputPCollection().getId() method.

Ditto for updating this in the batch pipeline translator as well.

As above, we need to pass a serializable representation to operators. We could create an ExecutableStage here and then reconstruct it on the runner. In that case, we need to ensure that two ExecutableStages constructed from the same ExecutableStagePayload are always equivalent (including any synthetic ids that may be generated). That should be the case as of now, but we need to be careful.

Unfortunately the way in which we construct coders and other properties of the ExecutableProcessBundleDescriptor are done during execution and it would be best if we somehow could make all these details be stable during pipeline translation so during execution it doesn't change. Having Flink rely on calling WireCoders.instantiateRunnerWireCoder(...) is an anti-pattern for encapsulation.

So if we need to construct the executable stage payload twice, we could make the contract have it be stable regardless the number of times it is constructed. I just want to push more of the input/output/coder/state/side input information upto translation time instead of having it deep within execution. In my opinion, only service (ApiServiceDescriptor) binding should happen there.

Unfortunately, Flink needs to have an associated serializer (TypeInformation, aka Coder) with each distributed collection. This TypeInformation needs to be known at pipeline construction time. It need not match the exact coder being used to materialize elements over gRPC, but it does need to match the in-memory element type.

We could get around this partially by representing everything as bytes. The downside is that each runner-native operation that requires structure (e.g., GBK) will require an additional operation to break elements into their constituent parts. This step itself also requires knowledge of the coded type, so we ultimately run into the same issue.

I'm just arguing for pushing most of the manipulation done within ExecutableProcessBundleDescriptor into the ExecutableStage payload (minus the ApiServiceDescriptor binding) so it doesn't need modification. This would allow the ExecutableStage to concretely answer what are the input coders, output coders, side input coders, state coders, ... in addition to any other information.

Longer term it seems if we had a way for the runner to say whether we need a keyed input context or grouped keyed output context makes sense as the runner could then say. These are the cases I know of:

KV<Key, Value> for SplittableDoFn input, StatefulDoFn input, GBK input, Multimap side input materialization input, window mapping input

KV<Key, Iterable> for GBK output

Do you know of any others?

I added a test for serialization. If we agree on the repeated instantiation of ExecutableStage, then I can take this up in a separate PR (for both, batch and streaming translation). I would do that once we have test and end-to-end coverage, right now the translators are still not wired.

lukecwik · 2018-05-23T17:40:05Z

...nk/src/main/java/org/apache/beam/runners/flink/FlinkStreamingPortablePipelineTranslator.java

-    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 =


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.

Removed, it was a remnant of translation code from old runner (only used in case of stateful ParDo). Since key handling and output encoding will happen in the SDK harness, we probably won't need to know the windowing strategy in the operator.

lukecwik · 2018-05-23T17:42:01Z

...nk/src/main/java/org/apache/beam/runners/flink/FlinkStreamingPortablePipelineTranslator.java

+    // we assume that the transformation does not change the windowing strategy.
+    RunnerApi.WindowingStrategy windowingStrategyProto =
+            pipeline.getComponents().getWindowingStrategiesOrThrow(
+                    pipeline.getComponents().getPcollectionsOrThrow(


You should rely on getInputPCollection().pcollection() from the ExecutableStage that you could construct above from the payload.

lukecwik · 2018-05-23T17:45:17Z

...nk/src/main/java/org/apache/beam/runners/flink/FlinkStreamingPortablePipelineTranslator.java

+    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);


Rely on creating an ExecutableStage here and its getOutputPCollections method. You can pass forward this ExecutableStage to the ExecutableStageDoFnOperator. Anywhere below where you need to get the input pcollection (and in the future state/timer or side input information), you can get it from this object.

I can see how having access to the coders would be important here and that the ExecutableStage should contain this information without needing to bind the data/state service that the ExecutableProcessBundleDescriptor does (currently that ExecutableProcessBundleDescriptor is munging the payload and making small albeit important modifications).

Unfortunately, we cannot simply pass the ExecutableStage onto operators here. Flink requires that operators be serializable. Operator constructors run on the client JVM, but operators are initialized via lifecycle methods on TaskManagers. For this reason, we use the executable stage payload in the batch translator. The same applies here.

We could decide to use a different serialized representation here for operator tasks, but it seems convenient here to reuse what we already have.

As commented above, we should be able to recreate the ExecutableStage multiple times and only bind the service (ApiServiceDescriptor) information should happen upstream.

lukecwik · 2018-05-23T17:46:20Z

.../main/java/org/apache/beam/runners/flink/translation/utils/FlinkPipelineTranslatorUtils.java

+  public static BiMap<String, Integer> createOutputMap(Iterable<String> localOutputs) {
+    ImmutableBiMap.Builder<String, Integer> builder = ImmutableBiMap.builder();
+    int outputIndex = 0;
+    for (String tag : localOutputs) {


Sort the localOutputs to get a stable indexing otherwise multiple calls to createOutputMap won't be stable.

lukecwik · 2018-05-23T18:08:36Z

...k/src/test/java/org/apache/beam/runners/flink/streaming/ExecutableStageDoFnOperatorTest.java

+  @Mock private RuntimeContext runtimeContext;
+  @Mock private DistributedCache distributedCache;
+  @Mock private FlinkExecutableStageContext stageContext;
+  @Mock private StageBundleFactory stageBundleFactory;


It might be easier to follow the testing strategy employed here:

beam/runners/java-fn-execution/src/test/java/org/apache/beam/runners/fnexecution/control/RemoteExecutionTest.java

Line 159 in 8345991

public void testExecution() throws Exception {

and use the InProcessSdkHarness TestRule to setup the tests instead of mocks.

If not, can review the tests as is.

I think the mock based test is good for covering just the operator class, without other dependencies. InProcessServerFactory might be a good way to write an integration test that also covers the translator, outside of the validate runner suite. I can probably do that as follow-up.

…ortable Flink runner.

tweise requested review from aljoscha and lukecwik May 18, 2018 02:37

tweise force-pushed the BEAM-4297.flinkStreamingExecutableStage branch 4 times, most recently from f1127a0 to 6c8c1c7 Compare May 18, 2018 05:36

bsidhom suggested changes May 19, 2018

View reviewed changes

tweise mentioned this pull request May 21, 2018

[BEAM-4271] Support side inputs for ExecutableStage and provide runner side utilities for handling multimap side inputs. #5374

Merged

10 tasks

tweise force-pushed the BEAM-4297.flinkStreamingExecutableStage branch from 720d134 to abeb4f2 Compare May 23, 2018 05:50

lukecwik requested changes May 23, 2018

View reviewed changes

tweise force-pushed the BEAM-4297.flinkStreamingExecutableStage branch 3 times, most recently from e44f783 to 424ed38 Compare May 29, 2018 16:23

lukecwik approved these changes May 29, 2018

View reviewed changes

[BEAM-4297] Streaming executable stage translation and operator for p…

6ed4d55

…ortable Flink runner.

tweise force-pushed the BEAM-4297.flinkStreamingExecutableStage branch from 424ed38 to 540d36e Compare May 29, 2018 17:37

[BEAM-4297] Add serialization test and make output mapping stable.

c6a0bf3

tweise force-pushed the BEAM-4297.flinkStreamingExecutableStage branch from 540d36e to c6a0bf3 Compare May 29, 2018 18:13

tweise merged commit c63de03 into apache:master May 29, 2018

Conversation

tweise commented May 18, 2018

Uh oh!

tweise commented May 18, 2018

Uh oh!

Choose a reason for hiding this comment

Uh oh!

tweise commented May 23, 2018

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

lukecwik May 23, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

lukecwik May 23, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

lukecwik May 23, 2018 •

edited

Loading

lukecwik May 23, 2018 •

edited

Loading