[BEAM-2898] Support Impulse transforms in Flink batch runner #4783

bsidhom · 2018-03-01T23:45:21Z

Fusion will be done via GreedyPipelineFuser, which does not currently support sourceless Read transforms. Instead, reads must happen via Impulse followed by a ParDo.

This change adds impulse support to the batch runner and a new ValidatesRunner ImpulseTest.

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

bsidhom · 2018-03-01T23:46:17Z

This depends on #477, #4717, and #4776 and should be submitted after those.

bsidhom · 2018-03-01T23:46:35Z

R: @tgroh

tgroh · 2018-03-01T23:53:25Z

...ruction-java/src/main/java/org/apache/beam/runners/core/construction/JavaReadViaImpulse.java

+   * safely use an unwrapped SerializableCoder because
+   * {@link SerializableCoder#structuralValue(Serializable)} assumes that coded elements support
+   * object equality. By default, Coders compare equality by serialized bytes, which we want in
+   * this case. It is usually safe to depend on coded representation here because we only compare


"usually safe..." isn't really necessary.

Probably worth noting that we usually expect Source equals methods to not be defined.

tgroh · 2018-03-01T23:54:08Z

...-java/src/main/java/org/apache/beam/runners/core/construction/graph/GreedyPipelineFuser.java

+  /**
+   * Fuses a {@link Pipeline} into a collection of {@link ExecutableStage}s.
+   *
+   * <p>This fuser expects each PTransform to have exactly one input. This means that pipelines must


Not true; Flatten is permitted, as is ParDo with side inputs.

However, it does expect all transforms with no inputs to not have an associated environment, which is what forces this requirement.

tgroh · 2018-03-02T00:03:24Z

...ink/src/main/java/org/apache/beam/runners/flink/translation/wrappers/ImpulseInputFormat.java

+
+  @Override
+  public WindowedValue<byte[]> nextRecord(WindowedValue<byte[]> windowedValue) {
+    hasOutput = false;


This name really confused me for a while. You're using it like "This split/format has output within it, which has not been produced" and I read it as "This split/format has produced output", which, of course, mean the opposite things.

'containsOutput' or 'availableOutput', perhaps?

Yes, I actually struggled to come up with a succinct yet descriptive name. Apparently this failed in the end... ;)

Updated to availableOutput and added a comment for clarity.

tgroh · 2018-03-02T00:04:05Z

sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ImpulseTest.java

+  @Test
+  @Category({ValidatesRunner.class, UsesImpulse.class})
+  public void testImpulseRead() {
+    PCollection<Integer> result = p.apply(JavaReadViaImpulse.bounded(Source.of(1, 2, 3)));


why not just a ParDo?

Will pipeline construction always insert an Impulse in that case? Or do I need to insert one myself? I want to ensure that runners can actually handle Impulse proper nodes.

What I mean is just insert an Impulse then a ParDo after it, instead of wrapping a Source

OK, I've added another test with a direct Impulse -> ParDo. Let me know if this looks good and I'll delete the source-based test.

This seeks to work fine and tests what we want to test, so I've gone ahead and removed the source-based test.

bsidhom

I've added a raw Impulse test and rebased.

bsidhom · 2018-03-08T19:47:56Z

sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ImpulseTest.java

+  @Test
+  @Category({ValidatesRunner.class, UsesImpulse.class})
+  public void testImpulseRead() {
+    PCollection<Integer> result = p.apply(JavaReadViaImpulse.bounded(Source.of(1, 2, 3)));


OK, I've added another test with a direct Impulse -> ParDo. Let me know if this looks good and I'll delete the source-based test.

tgroh · 2018-03-15T18:51:08Z

You'll need to rebase

tgroh · 2018-03-08T00:52:41Z

...ruction-java/src/main/java/org/apache/beam/runners/core/construction/JavaReadViaImpulse.java

-   * object equality. By default, Coders compare equality by serialized bytes, which we want in
-   * this case. It is usually safe to depend on coded representation here because we only compare
-   * objects on bundle commit, which compares serializations of the same object instance.
+   * object equality. Sources in general do not support object equality. By default, Coders compare


Though with #4817 I'm no longer hugely concerned about the coder stuff. I wouldn't revert it, however - certainly, the StructuralValue is better, even if it's not going to be used anywhere relevant, and the comment seems sufficient for the time being.

tgroh · 2018-03-08T00:53:24Z

...-java/src/main/java/org/apache/beam/runners/core/construction/graph/GreedyPipelineFuser.java

-   * use Impulse/ParDo transformations rather than read nodes. The utilities in
-   * {@link org.apache.beam.runners.core.construction.JavaReadViaImpulse} can be used to translate
-   * non-compliant pipelines.
+   * <p>This fuser expects each PTransform which has no inputs to have an associated environment.


s/an associated environment/no associated environment

tgroh · 2018-03-08T00:55:09Z

...-java/src/main/java/org/apache/beam/runners/core/construction/graph/GreedyPipelineFuser.java

-   * {@link org.apache.beam.runners.core.construction.JavaReadViaImpulse} can be used to translate
-   * non-compliant pipelines.
+   * <p>This fuser expects each PTransform which has no inputs to have an associated environment.
+   * This means that pipelines must use Impulse/ParDo transformations rather than read nodes. The


"This means that pipelines must be rooted at Impulse, or other runner-executed primitive transforms, instead of primitive Read nodes."

tgroh · 2018-03-15T18:51:50Z

...-java/src/main/java/org/apache/beam/runners/core/construction/graph/GreedyPipelineFuser.java

@@ -69,6 +69,15 @@ private GreedyPipelineFuser(Pipeline p) {
    fusePipeline(groupSiblings(rootConsumers));
  }

+  /**
+   * Fuses a {@link Pipeline} into a collection of {@link ExecutableStage}s.


{@link ExecutableStage ExecutableStages} is our normal style.

tgroh · 2018-03-15T18:52:15Z

...-java/src/main/java/org/apache/beam/runners/core/construction/graph/GreedyPipelineFuser.java

+   * must be rooted at Impulse, or other runner-executed primitive transforms, instead of primitive
+   * Read nodes. The utilities in
+   * {@link org.apache.beam.runners.core.construction.JavaReadViaImpulse} can be used to translate
+   * non-compliant pipelines.


This does kind of have an associated TODO for unbounded reads; https://issues.apache.org/jira/browse/BEAM-3859 is the (just-authored) issue to link against.

'can be used to translate non-compliant pipelines -> can be used to convert bounded pipelines using the Read primitive.'

tgroh · 2018-03-15T18:58:08Z

...ion-java/src/test/java/org/apache/beam/runners/core/construction/JavaReadViaImpulseTest.java

@@ -106,6 +149,13 @@ public void testOutputCoder() {
        equalTo(BigEndianIntegerCoder.of()));
  }

+  private static void assertNotReadTransform(PTransform<?, ?> transform) {


This could be inlined into the anonymous subclass that you'er traversing the pipeline with

As above, this is already submitted, though I could change it here if you want.

tgroh · 2018-03-15T18:58:31Z

...ion-java/src/test/java/org/apache/beam/runners/core/construction/JavaReadViaImpulseTest.java

+    PCollection<Long> input = p.apply(Read.from(source));
+    PAssert.that(input).containsInAnyOrder(0L, 1L, 2L, 3L, 4L, 5L, 6L, 7L, 8L, 9L);
+    p.replaceAll(Collections.singletonList(JavaReadViaImpulse.boundedOverride()));
+    p.traverseTopologically(new Pipeline.PipelineVisitor() {


extend PipelineVisitor.Defaults, and get rid of enterPipeline, visitValue, and leave*

This is actually already in master and was just one of the dependent changes.

tgroh · 2018-03-15T18:59:58Z

...ink/src/main/java/org/apache/beam/runners/flink/translation/wrappers/ImpulseInputFormat.java

+
+  @Override
+  public WindowedValue<byte[]> nextRecord(WindowedValue<byte[]> windowedValue) {
+    availableOutput = false;


This could be a checkArgument before you mutate the state (minor preference, no strong feelings)

tgroh · 2018-03-15T19:01:20Z

sdks/java/core/build.gradle

@@ -66,6 +66,7 @@ dependencies {
  shadow library.java.junit
  shadow "org.tukaani:xz:1.5"
  shadowTest project(":model:fn-execution").sourceSets.test.output
+  shadowTest project(path: ":runners:core-construction-java", configuration: "shadow")


What introduced this edge?

This also can't be represented in maven, and right now that worries me (though hopefully less soon)

This isn't needed anymore. Thanks for pointing it out.

bsidhom · 2018-04-02T22:56:58Z

Review comments addressed. PTAL.

tgroh · 2018-04-11T23:56:39Z

run java precommit

bsidhom · 2018-04-12T22:42:44Z

run java precommit

bsidhom · 2018-04-13T16:54:32Z

It looks like the precommits that are actually being run are passing.

tgroh requested changes Mar 2, 2018

View reviewed changes

bsidhom force-pushed the impulse-flink branch from 14470e5 to 9e6478b Compare March 8, 2018 19:48

bsidhom commented Mar 8, 2018

View reviewed changes

bsidhom force-pushed the impulse-flink branch from c5c0a2f to 9590590 Compare March 8, 2018 20:35

tgroh reviewed Mar 15, 2018

View reviewed changes

bsidhom force-pushed the impulse-flink branch from ef67102 to ae2bf61 Compare April 2, 2018 22:55

bsidhom added 2 commits April 2, 2018 16:01

Support impulse transforms in Flink

a43ccc2

Add Impulse ValidatesRunner test

3fb2694

bsidhom force-pushed the impulse-flink branch from ae2bf61 to 3fb2694 Compare April 2, 2018 23:01

tgroh approved these changes Apr 11, 2018

View reviewed changes

tgroh merged commit 007553f into apache:master Apr 13, 2018

bsidhom deleted the impulse-flink branch April 13, 2018 19:01

[BEAM-2898] Support Impulse transforms in Flink batch runner #4783

[BEAM-2898] Support Impulse transforms in Flink batch runner #4783

Conversation

bsidhom commented Mar 1, 2018

bsidhom commented Mar 1, 2018

bsidhom commented Mar 1, 2018

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

bsidhom left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

tgroh commented Mar 15, 2018

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

bsidhom commented Apr 2, 2018

tgroh commented Apr 11, 2018

bsidhom commented Apr 12, 2018

bsidhom commented Apr 13, 2018