You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by da...@apache.org on 2016/04/16 02:23:57 UTC
[1/2] incubator-beam git commit: Switch the Default PipelineRunner
Repository: incubator-beam
Updated Branches:
refs/heads/master e9f1b579a -> 9039949d5
Switch the Default PipelineRunner
Use the InProcessPiplineRunner (pending rename) as the default runner.
The InProcessPipelineRunner implements the beam model, including support
for Unbounded PCollections.
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/757cb326
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/757cb326
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/757cb326
Branch: refs/heads/master
Commit: 757cb326b909ad62aea8c51183a83521adfd5a3a
Parents: e9f1b57
Author: Thomas Groh <tg...@google.com>
Authored: Fri Apr 8 10:20:56 2016 -0700
Committer: Thomas Groh <tg...@google.com>
Committed: Fri Apr 15 16:12:52 2016 -0700
----------------------------------------------------------------------
.../beam/sdk/options/PipelineOptions.java | 4 +-
.../ImmutabilityCheckingBundleFactory.java | 20 +++--
.../sdk/options/PipelineOptionsFactoryTest.java | 3 +-
.../beam/sdk/options/PipelineOptionsTest.java | 4 +-
.../beam/sdk/runners/TransformTreeTest.java | 79 ++++++++++----------
.../EncodabilityEnforcementFactoryTest.java | 2 +-
.../ImmutabilityCheckingBundleFactoryTest.java | 17 ++---
.../apache/beam/sdk/transforms/ParDoTest.java | 14 ++--
.../beam/sdk/transforms/WithKeysJava8Test.java | 3 +-
9 files changed, 68 insertions(+), 78 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/757cb326/sdks/java/core/src/main/java/org/apache/beam/sdk/options/PipelineOptions.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/options/PipelineOptions.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/options/PipelineOptions.java
index 17cf5b3..d87e396 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/options/PipelineOptions.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/options/PipelineOptions.java
@@ -21,8 +21,8 @@ import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.options.GoogleApiDebugOptions.GoogleApiTracer;
import org.apache.beam.sdk.options.ProxyInvocationHandler.Deserializer;
import org.apache.beam.sdk.options.ProxyInvocationHandler.Serializer;
-import org.apache.beam.sdk.runners.DirectPipelineRunner;
import org.apache.beam.sdk.runners.PipelineRunner;
+import org.apache.beam.sdk.runners.inprocess.InProcessPipelineRunner;
import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.DoFn.Context;
@@ -225,7 +225,7 @@ public interface PipelineOptions {
@Description("The pipeline runner that will be used to execute the pipeline. "
+ "For registered runners, the class name can be specified, otherwise the fully "
+ "qualified name needs to be specified.")
- @Default.Class(DirectPipelineRunner.class)
+ @Default.Class(InProcessPipelineRunner.class)
Class<? extends PipelineRunner<?>> getRunner();
void setRunner(Class<? extends PipelineRunner<?>> kls);
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/757cb326/sdks/java/core/src/main/java/org/apache/beam/sdk/runners/inprocess/ImmutabilityCheckingBundleFactory.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/runners/inprocess/ImmutabilityCheckingBundleFactory.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/runners/inprocess/ImmutabilityCheckingBundleFactory.java
index 0852269..bb3d501 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/runners/inprocess/ImmutabilityCheckingBundleFactory.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/runners/inprocess/ImmutabilityCheckingBundleFactory.java
@@ -28,7 +28,6 @@ import org.apache.beam.sdk.util.IllegalMutationException;
import org.apache.beam.sdk.util.MutationDetector;
import org.apache.beam.sdk.util.MutationDetectors;
import org.apache.beam.sdk.util.SerializableUtils;
-import org.apache.beam.sdk.util.UserCodeException;
import org.apache.beam.sdk.util.WindowedValue;
import org.apache.beam.sdk.values.PCollection;
@@ -113,17 +112,16 @@ class ImmutabilityCheckingBundleFactory implements BundleFactory {
try {
detector.verifyUnmodified();
} catch (IllegalMutationException exn) {
- throw UserCodeException.wrap(
- new IllegalMutationException(
- String.format(
- "PTransform %s mutated value %s after it was output (new value was %s)."
- + " Values must not be mutated in any way after being output.",
- underlying.getPCollection().getProducingTransformInternal().getFullName(),
- exn.getSavedValue(),
- exn.getNewValue()),
+ throw new IllegalMutationException(
+ String.format(
+ "PTransform %s mutated value %s after it was output (new value was %s)."
+ + " Values must not be mutated in any way after being output.",
+ underlying.getPCollection().getProducingTransformInternal().getFullName(),
exn.getSavedValue(),
- exn.getNewValue(),
- exn));
+ exn.getNewValue()),
+ exn.getSavedValue(),
+ exn.getNewValue(),
+ exn);
}
}
return underlying.commit(synchronizedProcessingTime);
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/757cb326/sdks/java/core/src/test/java/org/apache/beam/sdk/options/PipelineOptionsFactoryTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/options/PipelineOptionsFactoryTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/options/PipelineOptionsFactoryTest.java
index 62c6909..e2d4342 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/options/PipelineOptionsFactoryTest.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/options/PipelineOptionsFactoryTest.java
@@ -31,6 +31,7 @@ import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.PipelineResult;
import org.apache.beam.sdk.runners.DirectPipelineRunner;
import org.apache.beam.sdk.runners.PipelineRunner;
+import org.apache.beam.sdk.runners.inprocess.InProcessPipelineRunner;
import org.apache.beam.sdk.testing.ExpectedLogs;
import org.apache.beam.sdk.testing.RestoreSystemProperties;
@@ -60,7 +61,7 @@ import java.util.Set;
@RunWith(JUnit4.class)
public class PipelineOptionsFactoryTest {
private static final Class<? extends PipelineRunner<?>> DEFAULT_RUNNER_CLASS =
- DirectPipelineRunner.class;
+ InProcessPipelineRunner.class;
@Rule public ExpectedException expectedException = ExpectedException.none();
@Rule public TestRule restoreSystemProperties = new RestoreSystemProperties();
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/757cb326/sdks/java/core/src/test/java/org/apache/beam/sdk/options/PipelineOptionsTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/options/PipelineOptionsTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/options/PipelineOptionsTest.java
index dfda528..459272e 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/options/PipelineOptionsTest.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/options/PipelineOptionsTest.java
@@ -24,7 +24,7 @@ import static org.junit.Assert.assertNotSame;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
-import org.apache.beam.sdk.runners.DirectPipelineRunner;
+import org.apache.beam.sdk.runners.inprocess.InProcessPipelineRunner;
import com.google.common.collect.Lists;
import com.google.common.collect.Sets;
@@ -87,7 +87,7 @@ public class PipelineOptionsTest {
@Test
public void testDefaultRunnerIsSet() {
- assertEquals(DirectPipelineRunner.class, PipelineOptionsFactory.create().getRunner());
+ assertEquals(InProcessPipelineRunner.class, PipelineOptionsFactory.create().getRunner());
}
@Test
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/757cb326/sdks/java/core/src/test/java/org/apache/beam/sdk/runners/TransformTreeTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/runners/TransformTreeTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/runners/TransformTreeTest.java
index 7690d2b..a778a0d 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/runners/TransformTreeTest.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/runners/TransformTreeTest.java
@@ -128,45 +128,46 @@ public class TransformTreeTest {
final EnumSet<TransformsSeen> left =
EnumSet.noneOf(TransformsSeen.class);
- p.traverseTopologically(new Pipeline.PipelineVisitor() {
- @Override
- public void enterCompositeTransform(TransformTreeNode node) {
- PTransform<?, ?> transform = node.getTransform();
- if (transform instanceof Sample.SampleAny) {
- assertTrue(visited.add(TransformsSeen.SAMPLE_ANY));
- assertNotNull(node.getEnclosingNode());
- assertTrue(node.isCompositeNode());
- } else if (transform instanceof Write.Bound) {
- assertTrue(visited.add(TransformsSeen.WRITE));
- assertNotNull(node.getEnclosingNode());
- assertTrue(node.isCompositeNode());
- }
- assertThat(transform, not(instanceOf(Read.Bounded.class)));
- }
-
- @Override
- public void leaveCompositeTransform(TransformTreeNode node) {
- PTransform<?, ?> transform = node.getTransform();
- if (transform instanceof Sample.SampleAny) {
- assertTrue(left.add(TransformsSeen.SAMPLE_ANY));
- }
- }
-
- @Override
- public void visitTransform(TransformTreeNode node) {
- PTransform<?, ?> transform = node.getTransform();
- // Pick is a composite, should not be visited here.
- assertThat(transform, not(instanceOf(Sample.SampleAny.class)));
- assertThat(transform, not(instanceOf(Write.Bound.class)));
- if (transform instanceof Read.Bounded) {
- assertTrue(visited.add(TransformsSeen.READ));
- }
- }
-
- @Override
- public void visitValue(PValue value, TransformTreeNode producer) {
- }
- });
+ p.traverseTopologically(
+ new Pipeline.PipelineVisitor() {
+ @Override
+ public void enterCompositeTransform(TransformTreeNode node) {
+ PTransform<?, ?> transform = node.getTransform();
+ if (transform instanceof Sample.SampleAny) {
+ assertTrue(visited.add(TransformsSeen.SAMPLE_ANY));
+ assertNotNull(node.getEnclosingNode());
+ assertTrue(node.isCompositeNode());
+ } else if (transform instanceof Write.Bound) {
+ assertTrue(visited.add(TransformsSeen.WRITE));
+ assertNotNull(node.getEnclosingNode());
+ assertTrue(node.isCompositeNode());
+ }
+ assertThat(transform, not(instanceOf(Read.Bounded.class)));
+ }
+
+ @Override
+ public void leaveCompositeTransform(TransformTreeNode node) {
+ PTransform<?, ?> transform = node.getTransform();
+ if (transform instanceof Sample.SampleAny) {
+ assertTrue(left.add(TransformsSeen.SAMPLE_ANY));
+ }
+ }
+
+ @Override
+ public void visitTransform(TransformTreeNode node) {
+ PTransform<?, ?> transform = node.getTransform();
+ // Pick is a composite, should not be visited here.
+ assertThat(transform, not(instanceOf(Sample.SampleAny.class)));
+ assertThat(transform, not(instanceOf(Write.Bound.class)));
+ if (transform instanceof Read.Bounded
+ && node.getEnclosingNode().getTransform() instanceof TextIO.Read.Bound) {
+ assertTrue(visited.add(TransformsSeen.READ));
+ }
+ }
+
+ @Override
+ public void visitValue(PValue value, TransformTreeNode producer) {}
+ });
assertTrue(visited.equals(EnumSet.allOf(TransformsSeen.class)));
assertTrue(left.equals(EnumSet.of(TransformsSeen.SAMPLE_ANY)));
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/757cb326/sdks/java/core/src/test/java/org/apache/beam/sdk/runners/inprocess/EncodabilityEnforcementFactoryTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/runners/inprocess/EncodabilityEnforcementFactoryTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/runners/inprocess/EncodabilityEnforcementFactoryTest.java
index 7720589..b3a7d15 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/runners/inprocess/EncodabilityEnforcementFactoryTest.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/runners/inprocess/EncodabilityEnforcementFactoryTest.java
@@ -55,7 +55,7 @@ public class EncodabilityEnforcementFactoryTest {
public void encodeFailsThrows() {
TestPipeline p = TestPipeline.create();
PCollection<Record> unencodable =
- p.apply(Create.of(new Record()).withCoder(new RecordNoEncodeCoder()));
+ p.apply(Create.<Record>of().withCoder(new RecordNoEncodeCoder()));
AppliedPTransform<?, ?, ?> consumer =
unencodable.apply(Count.<Record>globally()).getProducingTransformInternal();
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/757cb326/sdks/java/core/src/test/java/org/apache/beam/sdk/runners/inprocess/ImmutabilityCheckingBundleFactoryTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/runners/inprocess/ImmutabilityCheckingBundleFactoryTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/runners/inprocess/ImmutabilityCheckingBundleFactoryTest.java
index 386eacc..06e71b8 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/runners/inprocess/ImmutabilityCheckingBundleFactoryTest.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/runners/inprocess/ImmutabilityCheckingBundleFactoryTest.java
@@ -18,7 +18,6 @@
package org.apache.beam.sdk.runners.inprocess;
import static org.hamcrest.Matchers.containsInAnyOrder;
-import static org.hamcrest.Matchers.isA;
import static org.junit.Assert.assertThat;
import org.apache.beam.sdk.coders.ByteArrayCoder;
@@ -31,7 +30,6 @@ import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.transforms.windowing.IntervalWindow;
import org.apache.beam.sdk.transforms.windowing.PaneInfo;
import org.apache.beam.sdk.util.IllegalMutationException;
-import org.apache.beam.sdk.util.UserCodeException;
import org.apache.beam.sdk.util.WindowedValue;
import org.apache.beam.sdk.values.PCollection;
@@ -163,10 +161,9 @@ public class ImmutabilityCheckingBundleFactoryTest {
root.add(WindowedValue.valueInGlobalWindow(array));
array[1] = 2;
- thrown.expect(UserCodeException.class);
- thrown.expectCause(isA(IllegalMutationException.class));
+ thrown.expect(IllegalMutationException.class);
thrown.expectMessage("Values must not be mutated in any way after being output");
- CommittedBundle<byte[]> committed = root.commit(Instant.now());
+ root.commit(Instant.now());
}
@Test
@@ -184,10 +181,9 @@ public class ImmutabilityCheckingBundleFactoryTest {
keyed.add(windowedArray);
array[0] = Byte.MAX_VALUE;
- thrown.expect(UserCodeException.class);
- thrown.expectCause(isA(IllegalMutationException.class));
+ thrown.expect(IllegalMutationException.class);
thrown.expectMessage("Values must not be mutated in any way after being output");
- CommittedBundle<byte[]> committed = keyed.commit(Instant.now());
+ keyed.commit(Instant.now());
}
@Test
@@ -205,10 +201,9 @@ public class ImmutabilityCheckingBundleFactoryTest {
intermediate.add(windowedArray);
array[2] = -3;
- thrown.expect(UserCodeException.class);
- thrown.expectCause(isA(IllegalMutationException.class));
+ thrown.expect(IllegalMutationException.class);
thrown.expectMessage("Values must not be mutated in any way after being output");
- CommittedBundle<byte[]> committed = intermediate.commit(Instant.now());
+ intermediate.commit(Instant.now());
}
private static class IdentityDoFn<T> extends DoFn<T, T> {
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/757cb326/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ParDoTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ParDoTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ParDoTest.java
index 44154e6..83e0f2c 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ParDoTest.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ParDoTest.java
@@ -26,7 +26,6 @@ import static org.apache.beam.sdk.util.StringUtils.byteArrayToJsonString;
import static org.apache.beam.sdk.util.StringUtils.jsonStringToByteArray;
import static org.hamcrest.Matchers.allOf;
-import static org.hamcrest.Matchers.isA;
import static org.hamcrest.collection.IsIterableContainingInAnyOrder.containsInAnyOrder;
import static org.hamcrest.collection.IsIterableContainingInOrder.contains;
import static org.hamcrest.core.AnyOf.anyOf;
@@ -36,7 +35,6 @@ import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertThat;
import org.apache.beam.sdk.Pipeline;
-import org.apache.beam.sdk.Pipeline.PipelineExecutionException;
import org.apache.beam.sdk.coders.AtomicCoder;
import org.apache.beam.sdk.coders.CoderException;
import org.apache.beam.sdk.coders.ListCoder;
@@ -1119,7 +1117,7 @@ public class ParDoTest implements Serializable {
input.apply(ParDo.of(new SideOutputDummyFn(sideOutputTag))
.withOutputTags(mainOutputTag, TupleTagList.of(sideOutputTag)));
- thrown.expect(PipelineExecutionException.class);
+ thrown.expect(IllegalStateException.class);
thrown.expectMessage("Unable to return a default Coder");
pipeline.run();
}
@@ -1422,8 +1420,7 @@ public class ParDoTest implements Serializable {
}
}));
- thrown.expect(PipelineExecutionException.class);
- thrown.expectCause(isA(IllegalMutationException.class));
+ thrown.expect(IllegalMutationException.class);
thrown.expectMessage("output");
thrown.expectMessage("must not be mutated");
pipeline.run();
@@ -1472,8 +1469,7 @@ public class ParDoTest implements Serializable {
}
}));
- thrown.expect(PipelineExecutionException.class);
- thrown.expectCause(isA(IllegalMutationException.class));
+ thrown.expect(IllegalMutationException.class);
thrown.expectMessage("output");
thrown.expectMessage("must not be mutated");
pipeline.run();
@@ -1499,7 +1495,7 @@ public class ParDoTest implements Serializable {
}));
thrown.expect(IllegalMutationException.class);
- thrown.expectMessage("input");
+ thrown.expectMessage("Input");
thrown.expectMessage("must not be mutated");
pipeline.run();
}
@@ -1523,7 +1519,7 @@ public class ParDoTest implements Serializable {
}));
thrown.expect(IllegalMutationException.class);
- thrown.expectMessage("input");
+ thrown.expectMessage("Input");
thrown.expectMessage("must not be mutated");
pipeline.run();
}
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/757cb326/sdks/java/java8tests/src/test/java/org/apache/beam/sdk/transforms/WithKeysJava8Test.java
----------------------------------------------------------------------
diff --git a/sdks/java/java8tests/src/test/java/org/apache/beam/sdk/transforms/WithKeysJava8Test.java b/sdks/java/java8tests/src/test/java/org/apache/beam/sdk/transforms/WithKeysJava8Test.java
index a0d1a63..1ffb147 100644
--- a/sdks/java/java8tests/src/test/java/org/apache/beam/sdk/transforms/WithKeysJava8Test.java
+++ b/sdks/java/java8tests/src/test/java/org/apache/beam/sdk/transforms/WithKeysJava8Test.java
@@ -17,7 +17,6 @@
*/
package org.apache.beam.sdk.transforms;
-import org.apache.beam.sdk.Pipeline.PipelineExecutionException;
import org.apache.beam.sdk.testing.PAssert;
import org.apache.beam.sdk.testing.RunnableOnService;
import org.apache.beam.sdk.testing.TestPipeline;
@@ -65,7 +64,7 @@ public class WithKeysJava8Test {
values.apply("ApplyKeysWithWithKeys", WithKeys.of((String s) -> Integer.valueOf(s)));
- thrown.expect(PipelineExecutionException.class);
+ thrown.expect(IllegalStateException.class);
thrown.expectMessage("Unable to return a default Coder for ApplyKeysWithWithKeys");
thrown.expectMessage("Cannot provide a coder for type variable K");
thrown.expectMessage("the actual type is unknown due to erasure.");
[2/2] incubator-beam git commit: This closes #178
Posted by da...@apache.org.
This closes #178
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/9039949d
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/9039949d
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/9039949d
Branch: refs/heads/master
Commit: 9039949d5f518fed84bc7cf7e08870e023b53951
Parents: e9f1b57 757cb32
Author: Davor Bonaci <da...@google.com>
Authored: Fri Apr 15 17:07:42 2016 -0700
Committer: Davor Bonaci <da...@google.com>
Committed: Fri Apr 15 17:07:42 2016 -0700
----------------------------------------------------------------------
.../beam/sdk/options/PipelineOptions.java | 4 +-
.../ImmutabilityCheckingBundleFactory.java | 20 +++--
.../sdk/options/PipelineOptionsFactoryTest.java | 3 +-
.../beam/sdk/options/PipelineOptionsTest.java | 4 +-
.../beam/sdk/runners/TransformTreeTest.java | 79 ++++++++++----------
.../EncodabilityEnforcementFactoryTest.java | 2 +-
.../ImmutabilityCheckingBundleFactoryTest.java | 17 ++---
.../apache/beam/sdk/transforms/ParDoTest.java | 14 ++--
.../beam/sdk/transforms/WithKeysJava8Test.java | 3 +-
9 files changed, 68 insertions(+), 78 deletions(-)
----------------------------------------------------------------------