You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by tg...@apache.org on 2017/02/14 21:12:38 UTC
[1/2] beam git commit: This closes #1973
Repository: beam
Updated Branches:
refs/heads/master bea101a44 -> 938a66ec3
This closes #1973
Project: http://git-wip-us.apache.org/repos/asf/beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/beam/commit/938a66ec
Tree: http://git-wip-us.apache.org/repos/asf/beam/tree/938a66ec
Diff: http://git-wip-us.apache.org/repos/asf/beam/diff/938a66ec
Branch: refs/heads/master
Commit: 938a66ec3bc6d78aa7cd1992038f68f90cc283af
Parents: bea101a 4bd6d83
Author: Thomas Groh <tg...@google.com>
Authored: Tue Feb 14 13:12:15 2017 -0800
Committer: Thomas Groh <tg...@google.com>
Committed: Tue Feb 14 13:12:15 2017 -0800
----------------------------------------------------------------------
.../beam/runners/core/PTransformMatchers.java | 83 +++++++
.../runners/core/PTransformMatchersTest.java | 224 ++++++++++++++++---
2 files changed, 276 insertions(+), 31 deletions(-)
----------------------------------------------------------------------
[2/2] beam git commit: Add ParDo Matchers to PTransformMatchers
Posted by tg...@apache.org.
Add ParDo Matchers to PTransformMatchers
These match Splittable ParDos and ParDos that use State and Timers.
Update tests to remove excess generic args.
Project: http://git-wip-us.apache.org/repos/asf/beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/beam/commit/4bd6d833
Tree: http://git-wip-us.apache.org/repos/asf/beam/tree/4bd6d833
Diff: http://git-wip-us.apache.org/repos/asf/beam/diff/4bd6d833
Branch: refs/heads/master
Commit: 4bd6d833d11fdf7f3ccb7ad8d97735ea2fddfc07
Parents: bea101a
Author: Thomas Groh <tg...@google.com>
Authored: Thu Feb 9 11:43:24 2017 -0800
Committer: Thomas Groh <tg...@google.com>
Committed: Tue Feb 14 13:12:15 2017 -0800
----------------------------------------------------------------------
.../beam/runners/core/PTransformMatchers.java | 83 +++++++
.../runners/core/PTransformMatchersTest.java | 224 ++++++++++++++++---
2 files changed, 276 insertions(+), 31 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/beam/blob/4bd6d833/runners/core-java/src/main/java/org/apache/beam/runners/core/PTransformMatchers.java
----------------------------------------------------------------------
diff --git a/runners/core-java/src/main/java/org/apache/beam/runners/core/PTransformMatchers.java b/runners/core-java/src/main/java/org/apache/beam/runners/core/PTransformMatchers.java
index 362e8dc..1d7e24e 100644
--- a/runners/core-java/src/main/java/org/apache/beam/runners/core/PTransformMatchers.java
+++ b/runners/core-java/src/main/java/org/apache/beam/runners/core/PTransformMatchers.java
@@ -21,7 +21,12 @@ import org.apache.beam.sdk.annotations.Experimental;
import org.apache.beam.sdk.annotations.Experimental.Kind;
import org.apache.beam.sdk.runners.PTransformMatcher;
import org.apache.beam.sdk.transforms.AppliedPTransform;
+import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.PTransform;
+import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.transforms.reflect.DoFnSignature;
+import org.apache.beam.sdk.transforms.reflect.DoFnSignature.ProcessElementMethod;
+import org.apache.beam.sdk.transforms.reflect.DoFnSignatures;
/**
* A {@link PTransformMatcher} that matches {@link PTransform PTransforms} based on the class of the
@@ -56,4 +61,82 @@ public class PTransformMatchers {
return application.getTransform().getClass().equals(clazz);
}
}
+
+ /**
+ * A {@link PTransformMatcher} that matches a {@link ParDo.Bound} containing a {@link DoFn} that
+ * is splittable, as signified by {@link ProcessElementMethod#isSplittable()}.
+ */
+ public static PTransformMatcher splittableParDoSingle() {
+ return new PTransformMatcher() {
+ @Override
+ public boolean matches(AppliedPTransform<?, ?, ?> application) {
+ PTransform<?, ?> transform = application.getTransform();
+ if (transform instanceof ParDo.Bound) {
+ DoFn<?, ?> fn = ((ParDo.Bound<?, ?>) transform).getFn();
+ DoFnSignature signature = DoFnSignatures.signatureForDoFn(fn);
+ return signature.processElement().isSplittable();
+ }
+ return false;
+ }
+ };
+ }
+
+ /**
+ * A {@link PTransformMatcher} that matches a {@link ParDo.Bound} containing a {@link DoFn} that
+ * uses state or timers, as specified by {@link DoFnSignature#usesState()} and
+ * {@link DoFnSignature#usesTimers()}.
+ */
+ public static PTransformMatcher stateOrTimerParDoSingle() {
+ return new PTransformMatcher() {
+ @Override
+ public boolean matches(AppliedPTransform<?, ?, ?> application) {
+ PTransform<?, ?> transform = application.getTransform();
+ if (transform instanceof ParDo.Bound) {
+ DoFn<?, ?> fn = ((ParDo.Bound<?, ?>) transform).getFn();
+ DoFnSignature signature = DoFnSignatures.signatureForDoFn(fn);
+ return signature.usesState() || signature.usesTimers();
+ }
+ return false;
+ }
+ };
+ }
+
+ /**
+ * A {@link PTransformMatcher} that matches a {@link ParDo.BoundMulti} containing a {@link DoFn}
+ * that is splittable, as signified by {@link ProcessElementMethod#isSplittable()}.
+ */
+ public static PTransformMatcher splittableParDoMulti() {
+ return new PTransformMatcher() {
+ @Override
+ public boolean matches(AppliedPTransform<?, ?, ?> application) {
+ PTransform<?, ?> transform = application.getTransform();
+ if (transform instanceof ParDo.BoundMulti) {
+ DoFn<?, ?> fn = ((ParDo.BoundMulti<?, ?>) transform).getFn();
+ DoFnSignature signature = DoFnSignatures.signatureForDoFn(fn);
+ return signature.processElement().isSplittable();
+ }
+ return false;
+ }
+ };
+ }
+
+ /**
+ * A {@link PTransformMatcher} that matches a {@link ParDo.BoundMulti} containing a {@link DoFn}
+ * that uses state or timers, as specified by {@link DoFnSignature#usesState()} and
+ * {@link DoFnSignature#usesTimers()}.
+ */
+ public static PTransformMatcher stateOrTimerParDoMulti() {
+ return new PTransformMatcher() {
+ @Override
+ public boolean matches(AppliedPTransform<?, ?, ?> application) {
+ PTransform<?, ?> transform = application.getTransform();
+ if (transform instanceof ParDo.BoundMulti) {
+ DoFn<?, ?> fn = ((ParDo.BoundMulti<?, ?>) transform).getFn();
+ DoFnSignature signature = DoFnSignatures.signatureForDoFn(fn);
+ return signature.usesState() || signature.usesTimers();
+ }
+ return false;
+ }
+ };
+ }
}
http://git-wip-us.apache.org/repos/asf/beam/blob/4bd6d833/runners/core-java/src/test/java/org/apache/beam/runners/core/PTransformMatchersTest.java
----------------------------------------------------------------------
diff --git a/runners/core-java/src/test/java/org/apache/beam/runners/core/PTransformMatchersTest.java b/runners/core-java/src/test/java/org/apache/beam/runners/core/PTransformMatchersTest.java
index fe0c449..83795f1 100644
--- a/runners/core-java/src/test/java/org/apache/beam/runners/core/PTransformMatchersTest.java
+++ b/runners/core-java/src/test/java/org/apache/beam/runners/core/PTransformMatchersTest.java
@@ -23,18 +23,33 @@ import static org.hamcrest.Matchers.is;
import static org.hamcrest.Matchers.not;
import static org.junit.Assert.assertThat;
+import com.google.common.base.MoreObjects;
import java.io.Serializable;
+import org.apache.beam.sdk.coders.VarIntCoder;
import org.apache.beam.sdk.runners.PTransformMatcher;
import org.apache.beam.sdk.testing.TestPipeline;
import org.apache.beam.sdk.transforms.AppliedPTransform;
-import org.apache.beam.sdk.transforms.Create;
import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.PTransform;
import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.transforms.splittabledofn.RestrictionTracker;
import org.apache.beam.sdk.transforms.windowing.GlobalWindows;
import org.apache.beam.sdk.transforms.windowing.Window;
+import org.apache.beam.sdk.util.TimeDomain;
+import org.apache.beam.sdk.util.Timer;
+import org.apache.beam.sdk.util.TimerSpec;
+import org.apache.beam.sdk.util.TimerSpecs;
+import org.apache.beam.sdk.util.WindowingStrategy;
+import org.apache.beam.sdk.util.state.StateSpec;
+import org.apache.beam.sdk.util.state.StateSpecs;
+import org.apache.beam.sdk.util.state.ValueState;
+import org.apache.beam.sdk.values.KV;
import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.PCollection.IsBounded;
+import org.apache.beam.sdk.values.TupleTag;
+import org.apache.beam.sdk.values.TupleTagList;
import org.hamcrest.Matchers;
+import org.joda.time.Duration;
import org.junit.Rule;
import org.junit.Test;
import org.junit.runner.RunWith;
@@ -48,47 +63,52 @@ public class PTransformMatchersTest implements Serializable {
@Rule
public transient TestPipeline p = TestPipeline.create().enableAbandonedNodeEnforcement(false);
+ /**
+ * Gets the {@link AppliedPTransform} that has a created {@code PCollection<KV<String, Integer>>}
+ * as input.
+ */
+ private AppliedPTransform<?, ?, ?> getAppliedTransform(PTransform pardo) {
+ PCollection<KV<String, Integer>> input =
+ PCollection.createPrimitiveOutputInternal(
+ p, WindowingStrategy.globalDefault(), IsBounded.BOUNDED);
+ PCollection<Integer> output =
+ PCollection.createPrimitiveOutputInternal(
+ p, WindowingStrategy.globalDefault(), IsBounded.BOUNDED);
+
+ return AppliedPTransform.of("pardo", input.expand(), output.expand(), pardo, p);
+ }
+
@Test
public void classEqualToMatchesSameClass() {
PTransformMatcher matcher = PTransformMatchers.classEqualTo(ParDo.Bound.class);
- PCollection<Integer> input = p.apply(Create.of(1));
- ParDo.Bound<Integer, Integer> pardo = ParDo.of(new DoFn<Integer, Integer>() {
- @ProcessElement
- public void doStuff(ProcessContext ctxt) {
- }
- });
- PCollection<Integer> output = input.apply(pardo);
-
AppliedPTransform<?, ?, ?> application =
- AppliedPTransform
- .<PCollection<Integer>, PCollection<Integer>,
- PTransform<? super PCollection<Integer>, PCollection<Integer>>>
- of("DoStuff", input.expand(), output.expand(), pardo, p);
+ getAppliedTransform(
+ ParDo.of(
+ new DoFn<KV<String, Integer>, Integer>() {
+ @ProcessElement
+ public void doStuff(ProcessContext ctxt) {}
+ }));
assertThat(matcher.matches(application), is(true));
}
@Test
- public void classEqualToMatchesSubClass() {
- class MyPTransform extends PTransform<PCollection<Integer>, PCollection<Integer>> {
+ public void classEqualToDoesNotMatchSubclass() {
+ class MyPTransform extends PTransform<PCollection<KV<String, Integer>>, PCollection<Integer>> {
@Override
- public PCollection<Integer> expand(PCollection<Integer> input) {
- return input;
+ public PCollection<Integer> expand(PCollection<KV<String, Integer>> input) {
+ return PCollection.createPrimitiveOutputInternal(
+ input.getPipeline(), input.getWindowingStrategy(), input.isBounded());
}
}
PTransformMatcher matcher = PTransformMatchers.classEqualTo(MyPTransform.class);
- PCollection<Integer> input = p.apply(Create.of(1));
MyPTransform subclass = new MyPTransform() {};
assertThat(subclass.getClass(), not(Matchers.<Class<?>>equalTo(MyPTransform.class)));
assertThat(subclass, instanceOf(MyPTransform.class));
- PCollection<Integer> output = input.apply(subclass);
AppliedPTransform<?, ?, ?> application =
- AppliedPTransform
- .<PCollection<Integer>, PCollection<Integer>,
- PTransform<PCollection<Integer>, PCollection<Integer>>>
- of("DoStuff", input.expand(), output.expand(), subclass, p);
+ getAppliedTransform(subclass);
assertThat(matcher.matches(application), is(false));
}
@@ -96,16 +116,158 @@ public class PTransformMatchersTest implements Serializable {
@Test
public void classEqualToDoesNotMatchUnrelatedClass() {
PTransformMatcher matcher = PTransformMatchers.classEqualTo(ParDo.Bound.class);
- PCollection<Integer> input = p.apply(Create.of(1));
- Window.Bound<Integer> window = Window.into(new GlobalWindows());
- PCollection<Integer> output = input.apply(window);
-
AppliedPTransform<?, ?, ?> application =
- AppliedPTransform
- .<PCollection<Integer>, PCollection<Integer>,
- PTransform<PCollection<Integer>, PCollection<Integer>>>
- of("DoStuff", input.expand(), output.expand(), window, p);
+ getAppliedTransform(Window.<KV<String, Integer>>into(new GlobalWindows()));
assertThat(matcher.matches(application), is(false));
}
+
+ private DoFn<KV<String, Integer>, Integer> doFn =
+ new DoFn<KV<String, Integer>, Integer>() {
+ @ProcessElement
+ public void simpleProcess(ProcessContext ctxt) {
+ ctxt.output(ctxt.element().getValue() + 1);
+ }
+ };
+ private abstract static class SomeTracker implements RestrictionTracker<Void> {}
+ private DoFn<KV<String, Integer>, Integer> splittableDoFn =
+ new DoFn<KV<String, Integer>, Integer>() {
+ @ProcessElement
+ public void processElement(ProcessContext context, SomeTracker tracker) {}
+
+ @GetInitialRestriction
+ public Void getInitialRestriction(KV<String, Integer> element) {
+ return null;
+ }
+
+ @NewTracker
+ public SomeTracker newTracker(Void restriction) {
+ return null;
+ }
+ };
+ private DoFn<KV<String, Integer>, Integer> doFnWithState =
+ new DoFn<KV<String, Integer>, Integer>() {
+ private final String stateId = "mystate";
+
+ @StateId(stateId)
+ private final StateSpec<Object, ValueState<Integer>> intState =
+ StateSpecs.value(VarIntCoder.of());
+
+ @ProcessElement
+ public void processElement(ProcessContext c, @StateId(stateId) ValueState<Integer> state) {
+ Integer currentValue = MoreObjects.firstNonNull(state.read(), 0);
+ c.output(currentValue);
+ state.write(currentValue + 1);
+ }
+ };
+ private DoFn<KV<String, Integer>, Integer> doFnWithTimers =
+ new DoFn<KV<String, Integer>, Integer>() {
+ private final String timerId = "myTimer";
+
+ @TimerId(timerId)
+ private final TimerSpec spec = TimerSpecs.timer(TimeDomain.EVENT_TIME);
+
+ @ProcessElement
+ public void processElement(ProcessContext context, @TimerId(timerId) Timer timer) {
+ timer.setForNowPlus(Duration.standardSeconds(1));
+ context.output(3);
+ }
+
+ @OnTimer(timerId)
+ public void onTimer(OnTimerContext context) {
+ context.output(42);
+ }
+ };
+
+ /**
+ * Demonstrates that a {@link ParDo.Bound} does not match any ParDo matcher.
+ */
+ @Test
+ public void parDoSingle() {
+ AppliedPTransform<?, ?, ?> parDoApplication = getAppliedTransform(ParDo.of(doFn));
+
+ assertThat(PTransformMatchers.splittableParDoMulti().matches(parDoApplication), is(false));
+ assertThat(PTransformMatchers.stateOrTimerParDoMulti().matches(parDoApplication), is(false));
+ assertThat(PTransformMatchers.splittableParDoSingle().matches(parDoApplication), is(false));
+ assertThat(PTransformMatchers.stateOrTimerParDoSingle().matches(parDoApplication), is(false));
+ }
+
+ @Test
+ public void parDoSingleSplittable() {
+ AppliedPTransform<?, ?, ?> parDoApplication = getAppliedTransform(ParDo.of(splittableDoFn));
+ assertThat(PTransformMatchers.splittableParDoSingle().matches(parDoApplication), is(true));
+
+ assertThat(PTransformMatchers.splittableParDoMulti().matches(parDoApplication), is(false));
+ assertThat(PTransformMatchers.stateOrTimerParDoMulti().matches(parDoApplication), is(false));
+ assertThat(PTransformMatchers.stateOrTimerParDoSingle().matches(parDoApplication), is(false));
+ }
+
+ @Test
+ public void parDoSingleWithState() {
+ AppliedPTransform<?, ?, ?> parDoApplication = getAppliedTransform(ParDo.of(doFnWithState));
+ assertThat(PTransformMatchers.stateOrTimerParDoSingle().matches(parDoApplication), is(true));
+
+ assertThat(PTransformMatchers.splittableParDoMulti().matches(parDoApplication), is(false));
+ assertThat(PTransformMatchers.stateOrTimerParDoMulti().matches(parDoApplication), is(false));
+ assertThat(PTransformMatchers.splittableParDoSingle().matches(parDoApplication), is(false));
+ }
+
+ @Test
+ public void parDoSingleWithTimers() {
+ AppliedPTransform<?, ?, ?> parDoApplication =
+ getAppliedTransform(ParDo.of(doFnWithTimers));
+ assertThat(PTransformMatchers.stateOrTimerParDoSingle().matches(parDoApplication), is(true));
+
+ assertThat(PTransformMatchers.splittableParDoMulti().matches(parDoApplication), is(false));
+ assertThat(PTransformMatchers.stateOrTimerParDoMulti().matches(parDoApplication), is(false));
+ assertThat(PTransformMatchers.splittableParDoSingle().matches(parDoApplication), is(false));
+ }
+
+ @Test
+ public void parDoMulti() {
+ AppliedPTransform<?, ?, ?> parDoApplication =
+ getAppliedTransform(
+ ParDo.of(doFn).withOutputTags(new TupleTag<Integer>(), TupleTagList.empty()));
+
+ assertThat(PTransformMatchers.splittableParDoMulti().matches(parDoApplication), is(false));
+ assertThat(PTransformMatchers.stateOrTimerParDoMulti().matches(parDoApplication), is(false));
+ assertThat(PTransformMatchers.splittableParDoSingle().matches(parDoApplication), is(false));
+ assertThat(PTransformMatchers.stateOrTimerParDoSingle().matches(parDoApplication), is(false));
+ }
+
+ @Test
+ public void parDoMultiSplittable() {
+ AppliedPTransform<?, ?, ?> parDoApplication =
+ getAppliedTransform(
+ ParDo.of(splittableDoFn).withOutputTags(new TupleTag<Integer>(), TupleTagList.empty()));
+ assertThat(PTransformMatchers.splittableParDoMulti().matches(parDoApplication), is(true));
+
+ assertThat(PTransformMatchers.stateOrTimerParDoMulti().matches(parDoApplication), is(false));
+ assertThat(PTransformMatchers.splittableParDoSingle().matches(parDoApplication), is(false));
+ assertThat(PTransformMatchers.stateOrTimerParDoSingle().matches(parDoApplication), is(false));
+ }
+
+ @Test
+ public void parDoMultiWithState() {
+ AppliedPTransform<?, ?, ?> parDoApplication =
+ getAppliedTransform(
+ ParDo.of(doFnWithState).withOutputTags(new TupleTag<Integer>(), TupleTagList.empty()));
+ assertThat(PTransformMatchers.stateOrTimerParDoMulti().matches(parDoApplication), is(true));
+
+ assertThat(PTransformMatchers.splittableParDoMulti().matches(parDoApplication), is(false));
+ assertThat(PTransformMatchers.splittableParDoSingle().matches(parDoApplication), is(false));
+ assertThat(PTransformMatchers.stateOrTimerParDoSingle().matches(parDoApplication), is(false));
+ }
+
+ @Test
+ public void parDoMultiWithTimers() {
+ AppliedPTransform<?, ?, ?> parDoApplication =
+ getAppliedTransform(
+ ParDo.of(doFnWithTimers).withOutputTags(new TupleTag<Integer>(), TupleTagList.empty()));
+ assertThat(PTransformMatchers.stateOrTimerParDoMulti().matches(parDoApplication), is(true));
+
+ assertThat(PTransformMatchers.splittableParDoMulti().matches(parDoApplication), is(false));
+ assertThat(PTransformMatchers.splittableParDoSingle().matches(parDoApplication), is(false));
+ assertThat(PTransformMatchers.stateOrTimerParDoSingle().matches(parDoApplication), is(false));
+ }
}