[BEAM-310] Add RootTransformEvaluatorFactory #1051

tgroh · 2016-10-05T00:15:19Z

Be sure to do all of the following to help us incorporate your contribution
quickly and easily:

Make sure the PR title is formatted like:
[BEAM-<Jira issue #>] Description of pull request
Make sure tests pass via mvn clean verify. (Even better, enable
Travis-CI on your fork and ensure the whole test matrix passes).
Replace <Jira issue #> in the title with the actual Jira issue
number, if there is one.
If this contribution is large, please file an Apache
Individual Contributor License Agreement.

Use for Root Transforms.

These transforms generate their own initial inputs, which the Evaluator
is responsible for providing back to them to generate elements from the
root PCollections.

Update ExecutorServiceParallelExecutor to schedule roots based on the
provided transforms.

Some tests which have become no longer relevant are deleted within this PR.

Use for Root Transforms. These transforms generate their own initial inputs, which the Evaluator is responsible for providing back to them to generate elements from the root PCollections. Update ExecutorServiceParallelExecutor to schedule roots based on the provided transforms.

This interface is no longer used. Instead, the runner ensures that bundles will be provided containing the appropriate input to the TestStreamEvaluatorFactory.

tgroh · 2016-10-05T20:47:27Z

R: @lukecwik

lukecwik

Lets talk about my comments about the interface hierarchy before we proceed further.

Note to self, review tests.

lukecwik · 2016-10-05T22:40:10Z

...rs/direct-java/src/main/java/org/apache/beam/runners/direct/BoundedReadEvaluatorFactory.java

  }

+  @Override
+  public void cleanup() {}


cleanup -> cleanUp

This is the existing naming. If it's really important to you, I'm happy doing it in a separate PR, but it's not really relevant to this one.

not important

lukecwik · 2016-10-05T22:42:03Z

...rs/direct-java/src/main/java/org/apache/beam/runners/direct/BoundedReadEvaluatorFactory.java

+    return createInitialSplits((AppliedPTransform) transform);
+  }
+
+  private <OutputT> Collection<CommittedBundle<?>> createInitialSplits(


The plan is to later replace this with actual splitting in a follow up PR?

lukecwik · 2016-10-05T22:47:08Z

...irect-java/src/main/java/org/apache/beam/runners/direct/ExecutorServiceParallelExecutor.java

-    rootNodes = ImmutableList.copyOf(roots);
+    for (AppliedPTransform<?, ?, ?> root : roots) {
+      ConcurrentLinkedQueue<CommittedBundle<?>> pending = new ConcurrentLinkedQueue<>();
+      pending.addAll(registry.getInitialInputs(root));


Are we really limited to a single root per bundle in the model or is this an implementation detail.
I'm imagining that with SDF that this could potentially be multiple "root sources" per bundle.

A root bundle has a single transform associated with it, and that transform generates initial inputs sufficient to produce outputs - for SDF, this would be producing the initial restriction, or set of such restrictions, which would then be provided to the downstream Fn

lukecwik · 2016-10-05T22:49:07Z

...irect-java/src/main/java/org/apache/beam/runners/direct/ExecutorServiceParallelExecutor.java

-            if (ExecutorState.ACTIVE == startingState || (ExecutorState.PROCESSING == startingState
-                && noWorkOutstanding)) {
+            if (ExecutorState.ACTIVE == startingState
+                || (ExecutorState.PROCESSING == startingState && noWorkOutstanding)) {


also move && noWork... to next line for readability

Please still make this change

Whoops, sorry. Done.

lukecwik · 2016-10-05T22:49:34Z

...irect-java/src/main/java/org/apache/beam/runners/direct/ExecutorServiceParallelExecutor.java

@@ -419,6 +434,7 @@ private void fireTimers() throws Exception {
              }
              KeyedWorkItem<?, Object> work =
                  KeyedWorkItems.timersWorkItem(keyTimers.getKey().getKey(), delivery);
+              LOG.warn("Delivering {} timers for {}", delivery.size(), keyTimers.getKey().getKey());


Shouldn't be there at all, actually.

lukecwik · 2016-10-05T22:55:16Z

runners/direct-java/src/main/java/org/apache/beam/runners/direct/FlattenEvaluatorFactory.java

@@ -31,21 +34,32 @@
 * The {@link DirectRunner} {@link TransformEvaluatorFactory} for the {@link Flatten}
 * {@link PTransform}.
 */
-class FlattenEvaluatorFactory implements TransformEvaluatorFactory {
+class FlattenEvaluatorFactory implements RootTransformEvaluatorFactory {


It might be better to split this into two implementations, one which is expected as part of a flatten with upstream transforms and one flatten which is not. Otherwise the methods that you have will have to be "dual" purposed and may cause issues about how they are used in the future.

For example cleanUp may need to do something for both the root version and something for the non-root version and then you may have to add switching logic to figure out which is which.

This would also allow you to guard against the usage of forApplication for something that never produced output.

This change removes all the logic associated with handling a nonexistent input (if the Flatten is a root)

lukecwik · 2016-10-05T23:03:35Z

.../direct-java/src/main/java/org/apache/beam/runners/direct/RootTransformEvaluatorFactory.java

+ * {@link Pipeline}. Provides a way to get initial inputs, which will cause the {@link PTransform}
+ * to produce all of the appropriate output.
+ */
+interface RootTransformEvaluatorFactory extends TransformEvaluatorFactory {


Should it extend?

It seems as though the purpose of RootTransformEvaluatorFactory and TransformEvaluatorFactory are completely different, for example with the Flatten PTransform with 0 pcollection inputs, you would want it to be just a RootTransformEvaluatorFactory and not a regular TransformEvaluatorFactory

It should not. I've split it into an independent interface. However, I'm (generally) keeping the input providers within the Evaluator Factory implementations, as the initial inputs have to match the type of the actual Evaluator.

lukecwik · 2016-10-05T23:06:18Z

...ers/direct-java/src/main/java/org/apache/beam/runners/direct/TransformEvaluatorRegistry.java

@@ -77,15 +77,33 @@ private TransformEvaluatorRegistry(
  }

  @Override
+  public Collection<CommittedBundle<?>> getInitialInputs(AppliedPTransform<?, ?, ?> transform) {


This is an example where splitting the two interfaces might be wise so that you have a mapping from certain root ptransforms to evaluators for roots, and this mapping for the non-root ones.

lukecwik

Minor comments and then LGTM

I like the usage of the static inner class to keep the code close together but with a defined boundary.

lukecwik · 2016-10-06T20:42:16Z

...rs/direct-java/src/main/java/org/apache/beam/runners/direct/BoundedReadEvaluatorFactory.java

+              .add(WindowedValue.valueInGlobalWindow(BoundedSourceShard.of(source)))
+              .commit(BoundedWindow.TIMESTAMP_MAX_VALUE));
+    }
+


remove extraneous new lines

lukecwik · 2016-10-06T20:42:47Z

runners/direct-java/src/main/java/org/apache/beam/runners/direct/EmptyInputProvider.java

+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+


remove extra new line

lukecwik · 2016-10-06T20:42:56Z

runners/direct-java/src/main/java/org/apache/beam/runners/direct/EmptyInputProvider.java

+import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
+
+/**
+ * Created by tgroh on 10/6/16.


Update comment

lukecwik · 2016-10-06T20:44:27Z

runners/direct-java/src/main/java/org/apache/beam/runners/direct/RootProviderRegistry.java

+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+


remove extra new line

lukecwik · 2016-10-06T20:44:34Z

runners/direct-java/src/main/java/org/apache/beam/runners/direct/RootProviderRegistry.java

+import org.apache.beam.sdk.transforms.AppliedPTransform;
+import org.apache.beam.sdk.transforms.Flatten.FlattenPCollectionList;
+import org.apache.beam.sdk.transforms.PTransform;
+/**


add new line

lukecwik · 2016-10-06T20:49:24Z

...irect-java/src/main/java/org/apache/beam/runners/direct/ExecutorServiceParallelExecutor.java

-            if (ExecutorState.ACTIVE == startingState || (ExecutorState.PROCESSING == startingState
-                && noWorkOutstanding)) {
+            if (ExecutorState.ACTIVE == startingState
+                || (ExecutorState.PROCESSING == startingState && noWorkOutstanding)) {


Please still make this change

lukecwik · 2016-10-06T22:13:48Z

LGTM, merging

tgroh changed the title ~~Add RootTransformEvaluatorFactory~~ [BEAM-310] Add RootTransformEvaluatorFactory Oct 5, 2016

tgroh force-pushed the initial_inputs branch from 5b0fa8a to a409bcf Compare October 5, 2016 17:40

tgroh added 2 commits October 5, 2016 13:41

Remove KeyedResourcePool

86ab5f0

This interface is no longer used. Instead, the runner ensures that bundles will be provided containing the appropriate input to the TestStreamEvaluatorFactory.

tgroh force-pushed the initial_inputs branch from a409bcf to 86ab5f0 Compare October 5, 2016 20:47

lukecwik reviewed Oct 5, 2016

View reviewed changes

fixup! Add RootTransformEvaluatorFactory

c03bb1a

lukecwik requested changes Oct 6, 2016

View reviewed changes

fixup! Remove KeyedResourcePool

0e4d674

asfgit closed this in ecbc641 Oct 6, 2016

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[BEAM-310] Add RootTransformEvaluatorFactory #1051

[BEAM-310] Add RootTransformEvaluatorFactory #1051

tgroh commented Oct 5, 2016 •

edited

tgroh commented Oct 5, 2016

lukecwik left a comment

lukecwik Oct 5, 2016

tgroh Oct 6, 2016

lukecwik Oct 6, 2016

lukecwik Oct 5, 2016

tgroh Oct 5, 2016

lukecwik Oct 5, 2016

tgroh Oct 5, 2016

lukecwik Oct 5, 2016

lukecwik Oct 6, 2016

tgroh Oct 6, 2016

lukecwik Oct 5, 2016

tgroh Oct 5, 2016

lukecwik Oct 5, 2016 •

edited

tgroh Oct 5, 2016

lukecwik Oct 5, 2016

tgroh Oct 6, 2016

lukecwik Oct 5, 2016

tgroh Oct 5, 2016

lukecwik left a comment

lukecwik Oct 6, 2016

tgroh Oct 6, 2016

lukecwik Oct 6, 2016

tgroh Oct 6, 2016

lukecwik Oct 6, 2016

tgroh Oct 6, 2016

lukecwik Oct 6, 2016

tgroh Oct 6, 2016

lukecwik Oct 6, 2016

tgroh Oct 6, 2016

lukecwik Oct 6, 2016

lukecwik commented Oct 6, 2016

[BEAM-310] Add RootTransformEvaluatorFactory #1051

[BEAM-310] Add RootTransformEvaluatorFactory #1051

Conversation

tgroh commented Oct 5, 2016 • edited

tgroh commented Oct 5, 2016

lukecwik 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

Choose a reason for hiding this comment

lukecwik Oct 5, 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

lukecwik 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

lukecwik commented Oct 6, 2016

tgroh commented Oct 5, 2016 •

edited

lukecwik Oct 5, 2016 •

edited