Liberates ReduceFnRunner from WindowingInternals, and lets WindowingInternals do windowed side outputs #1353

jkff · 2016-11-12T01:08:46Z

Introduces WindowingInternals.sideOutputWindowedValue (will be necessary for Splittable DoFn)
Implements it properly in all runners (required some minor refactoring in Spark and Flink ProcessContext implementations)
Introduces "OutputWindowedValue" interface and "SideInputAccess" interfaces, and uses them in ReduceFnRunner instead of directly using WindowingInternals.
Introduces adapters from WindowingInternals to these two interfaces, for use in OldDoFn contexts
Moves some StateContext functions into ReduceFnContextFactory, because they make more sense in runners-core than in sdk (because they are only invoked by different runners). The only remaining StateContexts function is nullContext, but I couldn't figure out an easy way to move it into runners-core and gave up (however in fact I'm not sure its current usages are correct at all...)

R: @kennknowles (for bulk of the code and as committer)
CC: @aljoscha @amitsela (for the minor refactorings in respective runners)

aljoscha · 2016-11-12T08:27:33Z

runners/core-java/src/main/java/org/apache/beam/runners/core/SideInputAccess.java

+/**
+ * Allows accessing the side inputs for a particular main input window.
+ */
+public interface SideInputAccess {


Isn't there already SideInputReader?

+1 seems to be the same thing.

Reminds me that SideInputReader really should be in runners-core but it is still in the SDK due to misuse in CombineFnRunners and thereabouts.

Done. It took some fiddling with WindowingInternals - note also that now SideInputReader is the only user of WindowingInternals.sideInput. I'll be able to move SideInputReader into runners-core after another PR that @peihe is working on.

aljoscha · 2016-11-12T08:32:39Z

The changes on the Flink side look good 👍

amitsela · 2016-11-12T09:49:45Z

runners/spark/src/main/java/org/apache/beam/runners/spark/translation/DoFnFunction.java

@@ -32,7 +32,7 @@
 import org.apache.beam.sdk.values.TupleTag;
 import org.apache.spark.Accumulator;
 import org.apache.spark.api.java.function.FlatMapFunction;
-
+import org.joda.time.Instant;


unused.
Occurs in MultiDoFnFunction as well.

amitsela · 2016-11-12T10:04:51Z

runners/spark/src/main/java/org/apache/beam/runners/spark/translation/SparkProcessContext.java

-  }
-
-  @Override
-  public <T> void sideOutputWithTimestamp(TupleTag<T> tupleTag, T t, Instant instant) {


If you're opting for enabling side output by default (leaving DoFnFunction to explicitly override with "Unsupported"), that's fine by me, but I guess DoFnFunction should override sideOutputWithTimestamp with an UnsupportedOperationException as well.

The base class delegates normal output() and sideOutput() on the ProcessContext to the internal abstract methods I introduced - outputWindowedValue and sideOutputWindowedValue (I renamed them and made them protected in response to your comments). sideOutputWindowedValue in DoFnFunction is overridden to throw an UnsupportedOperationException.

amitsela · 2016-11-12T10:05:48Z

runners/spark/src/main/java/org/apache/beam/runners/spark/translation/DoFnFunction.java

+          "sideOutput is an unsupported operation for doFunctions, use a "
+              + "MultiDoFunction instead.");
+    }
+


This is a complementing comment to say that sideOutputWithTimestamp should throw UnsupportedOperationException as well.

Hopefully the comment above addresses this too. sideOutputWithTimestamp is defined in the base class and delegates to sideOutputWindowedValue, which here is overridden to throw the exception.

amitsela · 2016-11-12T10:08:45Z

runners/spark/src/main/java/org/apache/beam/runners/spark/translation/SparkProcessContext.java

+  }
+
+  public abstract void output(WindowedValue<OutputT> output);
+  public abstract <T> void sideOutput(TupleTag<T> tag, WindowedValue<T> output);


white space between the abstracts ?

amitsela · 2016-11-12T10:15:00Z

It seems that neither Spark or Flink runners were actually built by Jenkins as Direct runner failed on "unused imports" in checkstyle.

I've added some comments for the changes in the Spark runner, and I've also built the PR branch locally (ignoring checkstyle) and executed the integrations tests for local-runnable-on-service-tests for both Spark and Flink (to avoid surprises in post-commit) and I'm happy to say they both passed.

kennknowles · 2016-11-12T21:22:57Z

runners/core-java/src/main/java/org/apache/beam/runners/core/ReduceFnContextFactory.java

@@ -496,4 +501,51 @@ public Timers timers() {
      return timers;
    }
  }
+
+  private static <W extends BoundedWindow> StateContext<W> stateContextFromComponents(


Why move these?

They are only used by ReduceFnContextFactory (that's why I made them private to this class), and they in any case belong more into runners-core than into SDK - especially now that this method takes SideInputReader, which I would like to move into runners-core.

jkff · 2016-11-14T23:15:06Z

Thanks all, PTAL!

amitsela · 2016-11-14T23:35:32Z

LGTM pending Jenkins/Travis.

kennknowles

This is quite good. My comments are just slightly more than nits.

kennknowles · 2016-11-15T04:49:10Z

runners/core-java/src/main/java/org/apache/beam/runners/core/ReduceFnContextFactory.java

+
+        @Override
+        public <T> T sideInput(PCollectionView<T> view) {
+          return sideInputReader.get(view, windowFn.getSideInputWindow(mainInputWindow));


We had some discussions, and yes this is right.

kennknowles · 2016-11-15T04:53:02Z

runners/core-java/src/main/java/org/apache/beam/runners/core/SimpleDoFnRunner.java

+            SideOutputT output,
+            Instant timestamp,
+            Collection<? extends BoundedWindow> windows,
+            PaneInfo pane) {}


Seems like this (and the above which is duplicates) should both throw. A DoFn is not permitted to output to a different window. Silently dropping data seems a cruel way to enforce this.

kennknowles · 2016-11-15T04:55:45Z

runners/core-java/src/test/java/org/apache/beam/runners/core/ReduceFnTester.java


    @Override
    public void outputWindowedValue(KV<String, OutputT> output, Instant timestamp,
-        Collection<? extends BoundedWindow> windows, PaneInfo pane) {
+                                    Collection<? extends BoundedWindow> windows, PaneInfo pane) {


Don't align. Use google-java-format.

kennknowles · 2016-11-15T04:56:18Z

runners/core-java/src/test/java/org/apache/beam/runners/core/ReduceFnTester.java

+        Instant timestamp,
+        Collection<? extends BoundedWindow> windows,
+        PaneInfo pane) {
+      throw new UnsupportedOperationException();


Actionable error message, please. This might be a checkState if it is unreachable.

kennknowles · 2016-11-15T04:57:07Z

runners/core-java/src/test/java/org/apache/beam/runners/core/ReduceFnTester.java

+
+    @Override
+    public <T> boolean contains(PCollectionView<T> view) {
+      throw new UnsupportedOperationException();


kennknowles · 2016-11-15T04:57:13Z

runners/core-java/src/test/java/org/apache/beam/runners/core/ReduceFnTester.java

+
+    @Override
+    public boolean isEmpty() {
+      throw new UnsupportedOperationException();


kennknowles · 2016-11-15T04:57:26Z

...ect-java/src/main/java/org/apache/beam/runners/direct/GroupAlsoByWindowEvaluatorFactory.java

+
+                @Override
+                public <T> boolean contains(PCollectionView<T> view) {
+                  throw new UnsupportedOperationException();


Can you add an informative message to this exception?

kennknowles · 2016-11-15T04:57:34Z

...ect-java/src/main/java/org/apache/beam/runners/direct/GroupAlsoByWindowEvaluatorFactory.java

+
+                @Override
+                public boolean isEmpty() {
+                  throw new UnsupportedOperationException();


Can you add an informative message to this exception?

kennknowles · 2016-11-15T04:58:39Z

...ect-java/src/main/java/org/apache/beam/runners/direct/GroupAlsoByWindowEvaluatorFactory.java

+        Instant timestamp,
+        Collection<? extends BoundedWindow> windows,
+        PaneInfo pane) {
+      throw new UnsupportedOperationException("Can't output to side outputs from a ReduceFn");


Let's call it GroupAlsoByWindow, since actually that is the criteria. If ReduceFn ever becomes a thing, it might be allowed to side output. In fact, stateful DoFn is basically ReduceFn.

kennknowles · 2016-11-15T05:01:52Z

Just to direct your attention: the Jenkins failure is checkstyle in various Flink files. It runs before tests, to fail fast. The changes here will almost certainly fail spectacularly if they are incorrect, so I am not trying to replace testing by my manual inspection.

jkff · 2016-11-16T19:18:21Z

Dang. Looks like I can't just move StateContext.createFromComponents away from the SDK.

The Jenkins failure is:

(546b2b14867c0a52): Exception: java.lang.NoSuchMethodError: org.apache.beam.sdk.util.state.StateContexts.createFromComponents(Lorg/apache/beam/sdk/options/PipelineOptions;Lorg/apache/beam/sdk/util/WindowingInternals;Lorg/apache/beam/sdk/transforms/windowing/BoundedWindow;)Lorg/apache/beam/sdk/util/state/StateContext;

Let's see what I can do about this...

jkff · 2016-11-17T01:23:16Z

Green, PTAL.

…ut versions

Makes WindowingInternals.sideInput take the side input window instead of main input window.

It must be temporarily restored for compatibility with current Dataflow worker in order for integration tests to pass.

kennknowles

LGTM with the exception of the missing error messages.

kennknowles · 2016-11-17T22:33:02Z

...ect-java/src/main/java/org/apache/beam/runners/direct/GroupAlsoByWindowEvaluatorFactory.java

+
+                @Override
+                public <T> boolean contains(PCollectionView<T> view) {
+                  throw new UnsupportedOperationException();


Can you add an informative message to this exception?

kennknowles · 2016-11-17T22:33:05Z

...ect-java/src/main/java/org/apache/beam/runners/direct/GroupAlsoByWindowEvaluatorFactory.java

+
+                @Override
+                public boolean isEmpty() {
+                  throw new UnsupportedOperationException();


Can you add an informative message to this exception?

jkff · 2016-11-17T23:36:37Z

Done.

jkff mentioned this pull request Nov 12, 2016

[BEAM-801, BEAM-787] Improvements to Splittable DoFn #1261

Closed

aljoscha reviewed Nov 12, 2016

View reviewed changes

amitsela reviewed Nov 12, 2016

View reviewed changes

kennknowles reviewed Nov 12, 2016

View reviewed changes

kennknowles requested changes Nov 15, 2016

View reviewed changes

jkff force-pushed the reducefn-windowing-internals branch 3 times, most recently from 4db5bb4 to 1325673 Compare November 16, 2016 00:03

jkff force-pushed the reducefn-windowing-internals branch from 765cc77 to e230c9d Compare November 16, 2016 22:51

jkff added 5 commits November 17, 2016 13:18

Liberates ReduceFnRunner from WindowingInternals

24cae56

Refactor FlinkProcessContext more cleanly into single- and multi-outp…

1fb1f7b

…ut versions

Refactor SparkProcessContext more cleanly into single- and multi-outp…

8243fcd

…ut versions

Replaces SideInputAccess with SideInputReader

90a0d0e

Makes WindowingInternals.sideInput take the side input window instead of main input window.

Restores StateContexts.stateContextFromComponents

a23ae20

It must be temporarily restored for compatibility with current Dataflow worker in order for integration tests to pass.

kennknowles requested changes Nov 17, 2016

View reviewed changes

Added messages to exceptions

c0623c1

jkff force-pushed the reducefn-windowing-internals branch from e230c9d to c0623c1 Compare November 17, 2016 23:36

asfgit merged commit c0623c1 into apache:master Nov 18, 2016

asfgit pushed a commit that referenced this pull request Nov 18, 2016

This closes #1353

4852d2e

jkff deleted the reducefn-windowing-internals branch November 18, 2016 16:16

Liberates ReduceFnRunner from WindowingInternals, and lets WindowingInternals do windowed side outputs #1353

Liberates ReduceFnRunner from WindowingInternals, and lets WindowingInternals do windowed side outputs #1353

Conversation

jkff commented Nov 12, 2016

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

aljoscha commented Nov 12, 2016

amitsela Nov 12, 2016 • edited

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

amitsela commented Nov 12, 2016

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jkff commented Nov 14, 2016

amitsela commented Nov 14, 2016

kennknowles left a 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

kennknowles commented Nov 15, 2016 • edited

jkff commented Nov 16, 2016

jkff commented Nov 17, 2016

kennknowles left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jkff commented Nov 17, 2016

amitsela Nov 12, 2016 •

edited

kennknowles commented Nov 15, 2016 •

edited