You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by bc...@apache.org on 2016/05/11 16:31:11 UTC
[4/9] incubator-beam git commit: Test utility for display data in a
pipeline runner
Test utility for display data in a pipeline runner
DisplayDataEvaluator is useful for validating how PTransform
display data is surfaced in the context of a Pipeline and runner.
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/c4b41800
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/c4b41800
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/c4b41800
Branch: refs/heads/master
Commit: c4b418002ab05ca98f06e33f6c3e30effb28f90d
Parents: acb0406
Author: Scott Wegner <sw...@google.com>
Authored: Tue May 3 09:28:21 2016 -0700
Committer: bchambers <bc...@google.com>
Committed: Tue May 10 16:21:27 2016 -0700
----------------------------------------------------------------------
.../DataflowDisplayDataEvaluator.java | 71 +++++++++
.../display/DisplayDataEvaluator.java | 146 +++++++++++++++++++
.../display/DisplayDataEvaluatorTest.java | 74 ++++++++++
3 files changed, 291 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/c4b41800/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/transforms/DataflowDisplayDataEvaluator.java
----------------------------------------------------------------------
diff --git a/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/transforms/DataflowDisplayDataEvaluator.java b/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/transforms/DataflowDisplayDataEvaluator.java
new file mode 100644
index 0000000..db42c77
--- /dev/null
+++ b/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/transforms/DataflowDisplayDataEvaluator.java
@@ -0,0 +1,71 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.runners.dataflow.transforms;
+
+import org.apache.beam.runners.dataflow.DataflowPipelineRunner;
+import org.apache.beam.runners.dataflow.options.DataflowPipelineDebugOptions;
+import org.apache.beam.runners.dataflow.options.DataflowPipelineOptions;
+import org.apache.beam.sdk.options.GcpOptions;
+import org.apache.beam.sdk.options.PipelineOptionsFactory;
+import org.apache.beam.sdk.transforms.display.DisplayDataEvaluator;
+import org.apache.beam.sdk.util.NoopCredentialFactory;
+import org.apache.beam.sdk.util.NoopPathValidator;
+
+import com.google.common.collect.Lists;
+
+/**
+ * Factory methods for creating {@link DisplayDataEvaluator} instances against the
+ * {@link DataflowPipelineRunner}.
+ */
+public final class DataflowDisplayDataEvaluator {
+ /** Do not instantiate. */
+ private DataflowDisplayDataEvaluator() {}
+
+ /**
+ * Retrieve a set of default {@link DataflowPipelineOptions} which can be used to build
+ * dataflow pipelines for evaluating display data.
+ */
+ public static DataflowPipelineOptions getDefaultOptions() {
+ DataflowPipelineOptions options = PipelineOptionsFactory.as(DataflowPipelineOptions.class);
+
+ options.setProject("foobar");
+ options.setTempLocation("gs://bucket/tmpLocation");
+ options.setFilesToStage(Lists.<String>newArrayList());
+
+ options.as(DataflowPipelineDebugOptions.class).setPathValidatorClass(NoopPathValidator.class);
+ options.as(GcpOptions.class).setCredentialFactoryClass(NoopCredentialFactory.class);
+
+ return options;
+ }
+
+ /**
+ * Create a {@link DisplayDataEvaluator} instance to evaluate pipeline display data against
+ * the {@link DataflowPipelineRunner}.
+ */
+ public static DisplayDataEvaluator create() {
+ return create(getDefaultOptions());
+ }
+
+ /**
+ * Create a {@link DisplayDataEvaluator} instance to evaluate pipeline display data against
+ * the {@link DataflowPipelineRunner} with the specified {@code options}.
+ */
+ public static DisplayDataEvaluator create(DataflowPipelineOptions options) {
+ return DisplayDataEvaluator.forRunner(DataflowPipelineRunner.class, options);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/c4b41800/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/display/DisplayDataEvaluator.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/display/DisplayDataEvaluator.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/display/DisplayDataEvaluator.java
new file mode 100644
index 0000000..6759003
--- /dev/null
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/display/DisplayDataEvaluator.java
@@ -0,0 +1,146 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.transforms.display;
+
+import org.apache.beam.sdk.Pipeline;
+import org.apache.beam.sdk.coders.Coder;
+import org.apache.beam.sdk.options.PipelineOptions;
+import org.apache.beam.sdk.options.PipelineOptionsFactory;
+import org.apache.beam.sdk.runners.PipelineRunner;
+import org.apache.beam.sdk.runners.TransformTreeNode;
+import org.apache.beam.sdk.transforms.Create;
+import org.apache.beam.sdk.transforms.PTransform;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.POutput;
+
+import com.google.common.collect.Sets;
+
+import java.util.Objects;
+import java.util.Set;
+
+/**
+ * Test utilities to evaluate the {@link DisplayData} in the context of a {@link PipelineRunner}.
+ */
+public class DisplayDataEvaluator {
+ private final PipelineOptions options;
+
+ /**
+ * Create a new {@link DisplayDataEvaluator} using the specified {@link PipelineRunner} and
+ * default {@link PipelineOptions}.
+ */
+ public static DisplayDataEvaluator forRunner(Class<? extends PipelineRunner<?>> pipelineRunner) {
+ return forRunner(pipelineRunner, PipelineOptionsFactory.create());
+ }
+
+ /**
+ * Create a new {@link DisplayDataEvaluator} using the specified {@link PipelineRunner} and
+ * {@link PipelineOptions}.
+ */
+ public static DisplayDataEvaluator forRunner(
+ Class<? extends PipelineRunner<?>> pipelineRunner, PipelineOptions pipelineOptions) {
+ return new DisplayDataEvaluator(pipelineRunner, pipelineOptions);
+ }
+
+ private DisplayDataEvaluator(Class<? extends PipelineRunner<?>> runner, PipelineOptions options) {
+ this.options = options;
+ this.options.setRunner(runner);
+ }
+
+ /**
+ * Traverse the specified {@link PTransform}, collecting {@link DisplayData} registered on the
+ * inner primitive {@link PTransform PTransforms}.
+ *
+ * @return the set of {@link DisplayData} for primitive {@link PTransform PTransforms}.
+ */
+ public <InputT> Set<DisplayData> displayDataForPrimitiveTransforms(
+ final PTransform<? super PCollection<InputT>, ? extends POutput> root) {
+ return displayDataForPrimitiveTransforms(root, null);
+ }
+
+ /**
+ * Traverse the specified {@link PTransform}, collecting {@link DisplayData} registered on the
+ * inner primitive {@link PTransform PTransforms}.
+ *
+ * @param root The root {@link PTransform} to traverse
+ * @param inputCoder The coder to set for the {@link PTransform} input, or null to infer the
+ * default coder.
+ *
+ * @return the set of {@link DisplayData} for primitive {@link PTransform PTransforms}.
+ */
+ public <InputT> Set<DisplayData> displayDataForPrimitiveTransforms(
+ final PTransform<? super PCollection<InputT>, ? extends POutput> root,
+ Coder<InputT> inputCoder) {
+
+ Create.Values<InputT> input = Create.of();
+ if (inputCoder != null) {
+ input = input.withCoder(inputCoder);
+ }
+
+ Pipeline pipeline = Pipeline.create(options);
+ pipeline
+ .apply(input)
+ .apply(root);
+
+ PrimitiveDisplayDataPTransformVisitor visitor = new PrimitiveDisplayDataPTransformVisitor(root);
+ pipeline.traverseTopologically(visitor);
+ return visitor.getPrimitivesDisplayData();
+ }
+
+ /**
+ * Visits {@link PTransform PTransforms} in a pipeline, and collects {@link DisplayData} for
+ * each primitive transform under a given composite root.
+ */
+ private static class PrimitiveDisplayDataPTransformVisitor
+ extends Pipeline.PipelineVisitor.Defaults {
+ private final PTransform root;
+ private final Set<DisplayData> displayData;
+ private boolean shouldRecord = false;
+
+ PrimitiveDisplayDataPTransformVisitor(PTransform root) {
+ this.root = root;
+ this.displayData = Sets.newHashSet();
+ }
+
+ Set<DisplayData> getPrimitivesDisplayData() {
+ return displayData;
+ }
+
+ @Override
+ public CompositeBehavior enterCompositeTransform(TransformTreeNode node) {
+ if (Objects.equals(root, node.getTransform())) {
+ shouldRecord = true;
+ }
+
+ return CompositeBehavior.ENTER_TRANSFORM;
+ }
+
+ @Override
+ public void leaveCompositeTransform(TransformTreeNode node) {
+ if (Objects.equals(root, node.getTransform())) {
+ shouldRecord = false;
+ }
+ }
+
+ @Override
+ public void visitPrimitiveTransform(TransformTreeNode node) {
+ if (shouldRecord) {
+ displayData.add(DisplayData.from(node.getTransform()));
+ }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/c4b41800/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/display/DisplayDataEvaluatorTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/display/DisplayDataEvaluatorTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/display/DisplayDataEvaluatorTest.java
new file mode 100644
index 0000000..f24c133
--- /dev/null
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/display/DisplayDataEvaluatorTest.java
@@ -0,0 +1,74 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.transforms.display;
+
+import static org.apache.beam.sdk.transforms.display.DisplayDataMatchers.hasDisplayItem;
+
+import static org.hamcrest.Matchers.hasItem;
+import static org.hamcrest.Matchers.not;
+import static org.junit.Assert.assertThat;
+
+import org.apache.beam.sdk.runners.DirectPipelineRunner;
+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.values.PCollection;
+import org.apache.beam.sdk.values.POutput;
+
+import org.junit.Test;
+
+import java.io.Serializable;
+import java.util.Set;
+
+/**
+ * Unit tests for {@link DisplayDataEvaluator}.
+ */
+public class DisplayDataEvaluatorTest implements Serializable {
+
+ @Test
+ public void testDisplayDataForPrimitiveTransforms() {
+ PTransform<? super PCollection<String>, ? super POutput> myTransform =
+ new PTransform<PCollection<String>, POutput> () {
+ @Override
+ public PCollection<String> apply(PCollection<String> input) {
+ return input.apply(ParDo.of(new DoFn<String, String>() {
+ @Override
+ public void processElement(ProcessContext c) throws Exception {
+ c.output(c.element());
+ }
+
+ @Override
+ public void populateDisplayData(DisplayData.Builder builder) {
+ builder.add(DisplayData.item("primitiveKey", "primitiveValue"));
+ }
+ }));
+ }
+
+ @Override
+ public void populateDisplayData(DisplayData.Builder builder) {
+ builder.add(DisplayData.item("compositeKey", "compositeValue"));
+ }
+ };
+
+ DisplayDataEvaluator evaluator = DisplayDataEvaluator.forRunner(DirectPipelineRunner.class);
+ Set<DisplayData> displayData = evaluator.displayDataForPrimitiveTransforms(myTransform);
+
+ assertThat(displayData, not(hasItem(hasDisplayItem("compositeKey", "compositeValue"))));
+ assertThat(displayData, hasItem(hasDisplayItem("primitiveKey", "primitiveValue")));
+ }
+}