Add Initial Materializations
Project: http://git-wip-us.apache.org/repos/asf/beam/repo Commit: http://git-wip-us.apache.org/repos/asf/beam/commit/9b5d66ee Tree: http://git-wip-us.apache.org/repos/asf/beam/tree/9b5d66ee Diff: http://git-wip-us.apache.org/repos/asf/beam/diff/9b5d66ee Branch: refs/heads/master Commit: 9b5d66eec14501769086a12347960bdf52d62921 Parents: 447fb2c Author: Thomas Groh <tg...@google.com> Authored: Mon Apr 3 16:25:08 2017 -0700 Committer: Thomas Groh <tg...@google.com> Committed: Wed Apr 5 18:09:27 2017 -0700 ---------------------------------------------------------------------- .../beam/sdk/transforms/Materialization.java | 36 +++++++++++++ .../beam/sdk/transforms/Materializations.java | 53 ++++++++++++++++++++ .../org/apache/beam/sdk/transforms/ViewFn.java | 6 +++ .../apache/beam/sdk/util/PCollectionViews.java | 25 +++++++++ .../sdk/testing/PCollectionViewTesting.java | 18 +++++++ 5 files changed, 138 insertions(+) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/beam/blob/9b5d66ee/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Materialization.java ---------------------------------------------------------------------- diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Materialization.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Materialization.java new file mode 100644 index 0000000..7cd6256 --- /dev/null +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Materialization.java @@ -0,0 +1,36 @@ +/* + * 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; + +import org.apache.beam.sdk.runners.PipelineRunner; + +/** + * How a view should be physically materialized by a {@link PipelineRunner}. + * + * <p>A {@link PipelineRunner} will support some set of materializations, and will reject + * {@link ViewFn ViewFns} that require materializations it does not support. See + * {@link Materializations} for known implementations. + */ +public interface Materialization<T> { + /** + * Gets the URN describing this {@link Materialization}. This is a stable, SDK-independent URN + * understood by a {@link PipelineRunner}. + */ + String getUrn(); +} http://git-wip-us.apache.org/repos/asf/beam/blob/9b5d66ee/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Materializations.java ---------------------------------------------------------------------- diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Materializations.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Materializations.java new file mode 100644 index 0000000..35925fa --- /dev/null +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Materializations.java @@ -0,0 +1,53 @@ +/* + * 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; + +import org.apache.beam.sdk.annotations.Experimental; +import org.apache.beam.sdk.annotations.Experimental.Kind; +import org.apache.beam.sdk.util.WindowedValue; + +/** + * Utility methods for constructing known {@link Materialization materializations} for a + * {@link ViewFn}. + */ +public class Materializations { + /** + * The URN for a {@link Materialization} where the primitive view type is an iterable of fully + * specified windowed values. + */ + @Experimental(Kind.CORE_RUNNERS_ONLY) + public static final String ITERABLE_MATERIALIZATION_URN = + "urn:beam:sideinput:materialization:iterable:0.1"; + + /** + * A {@link Materialization} where the primitive view type is an iterable of fully specified + * windowed values. + */ + public static <T> Materialization<Iterable<WindowedValue<T>>> iterable() { + return new IterableMaterialization<>(); + } + + private static class IterableMaterialization<T> + implements Materialization<Iterable<WindowedValue<T>>> { + @Override + public String getUrn() { + return ITERABLE_MATERIALIZATION_URN; + } + } +} http://git-wip-us.apache.org/repos/asf/beam/blob/9b5d66ee/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/ViewFn.java ---------------------------------------------------------------------- diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/ViewFn.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/ViewFn.java index 981d047..ff53d1c 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/ViewFn.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/ViewFn.java @@ -18,6 +18,7 @@ package org.apache.beam.sdk.transforms; import java.io.Serializable; +import org.apache.beam.sdk.runners.PipelineRunner; import org.apache.beam.sdk.values.PCollection; import org.apache.beam.sdk.values.PCollectionView; @@ -38,6 +39,11 @@ import org.apache.beam.sdk.values.PCollectionView; */ public abstract class ViewFn<PrimitiveViewT, ViewT> implements Serializable { /** + * Gets the materialization of this {@link ViewFn}. + */ + public abstract Materialization<PrimitiveViewT> getMaterialization(); + + /** * A function to adapt a primitive view type to a desired view type. */ public abstract ViewT apply(PrimitiveViewT contents); http://git-wip-us.apache.org/repos/asf/beam/blob/9b5d66ee/sdks/java/core/src/main/java/org/apache/beam/sdk/util/PCollectionViews.java ---------------------------------------------------------------------- diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/PCollectionViews.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/util/PCollectionViews.java index 7617253..14b36fd 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/PCollectionViews.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/util/PCollectionViews.java @@ -35,6 +35,8 @@ import org.apache.beam.sdk.annotations.Experimental; import org.apache.beam.sdk.annotations.Experimental.Kind; import org.apache.beam.sdk.coders.Coder; import org.apache.beam.sdk.coders.IterableCoder; +import org.apache.beam.sdk.transforms.Materialization; +import org.apache.beam.sdk.transforms.Materializations; import org.apache.beam.sdk.transforms.ViewFn; import org.apache.beam.sdk.transforms.windowing.BoundedWindow; import org.apache.beam.sdk.transforms.windowing.InvalidWindows; @@ -193,6 +195,11 @@ public class PCollectionViews { } @Override + public Materialization<Iterable<WindowedValue<T>>> getMaterialization() { + return Materializations.iterable(); + } + + @Override public T apply(Iterable<WindowedValue<T>> contents) { try { return Iterables.getOnlyElement(contents).getValue(); @@ -220,6 +227,10 @@ public class PCollectionViews { @Experimental(Kind.CORE_RUNNERS_ONLY) public static class IterableViewFn<T> extends ViewFn<Iterable<WindowedValue<T>>, Iterable<T>> { + @Override + public Materialization<Iterable<WindowedValue<T>>> getMaterialization() { + return Materializations.iterable(); + } @Override public Iterable<T> apply(Iterable<WindowedValue<T>> contents) { @@ -248,6 +259,11 @@ public class PCollectionViews { @Experimental(Kind.CORE_RUNNERS_ONLY) public static class ListViewFn<T> extends ViewFn<Iterable<WindowedValue<T>>, List<T>> { @Override + public Materialization<Iterable<WindowedValue<T>>> getMaterialization() { + return Materializations.iterable(); + } + + @Override public List<T> apply(Iterable<WindowedValue<T>> contents) { return ImmutableList.copyOf( Iterables.transform(contents, new Function<WindowedValue<T>, T>() { @@ -271,6 +287,10 @@ public class PCollectionViews { @Experimental(Kind.CORE_RUNNERS_ONLY) public static class MultimapViewFn<K, V> extends ViewFn<Iterable<WindowedValue<KV<K, V>>>, Map<K, Iterable<V>>> { + @Override + public Materialization<Iterable<WindowedValue<KV<K, V>>>> getMaterialization() { + return Materializations.iterable(); + } @Override public Map<K, Iterable<V>> apply(Iterable<WindowedValue<KV<K, V>>> elements) { @@ -296,6 +316,11 @@ public class PCollectionViews { @Deprecated @Experimental(Kind.CORE_RUNNERS_ONLY) public static class MapViewFn<K, V> extends ViewFn<Iterable<WindowedValue<KV<K, V>>>, Map<K, V>> { + @Override + public Materialization<Iterable<WindowedValue<KV<K, V>>>> getMaterialization() { + return Materializations.iterable(); + } + /** * Input iterable must actually be {@code Iterable<WindowedValue<KV<K, V>>>}. */ http://git-wip-us.apache.org/repos/asf/beam/blob/9b5d66ee/sdks/java/core/src/test/java/org/apache/beam/sdk/testing/PCollectionViewTesting.java ---------------------------------------------------------------------- diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/testing/PCollectionViewTesting.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/testing/PCollectionViewTesting.java index e6b13c1..5f71cab 100644 --- a/sdks/java/core/src/test/java/org/apache/beam/sdk/testing/PCollectionViewTesting.java +++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/testing/PCollectionViewTesting.java @@ -15,6 +15,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ + package org.apache.beam.sdk.testing; import com.google.common.base.Function; @@ -25,6 +26,8 @@ import java.util.List; import java.util.Objects; import org.apache.beam.sdk.coders.Coder; import org.apache.beam.sdk.coders.IterableCoder; +import org.apache.beam.sdk.transforms.Materialization; +import org.apache.beam.sdk.transforms.Materializations; import org.apache.beam.sdk.transforms.ViewFn; import org.apache.beam.sdk.transforms.windowing.BoundedWindow; import org.apache.beam.sdk.transforms.windowing.FixedWindows; @@ -86,6 +89,11 @@ public final class PCollectionViewTesting { */ public static class IdentityViewFn<T> extends ViewFn<Iterable<WindowedValue<T>>, Iterable<T>> { @Override + public Materialization<Iterable<WindowedValue<T>>> getMaterialization() { + return Materializations.iterable(); + } + + @Override public Iterable<T> apply(Iterable<WindowedValue<T>> contents) { return Iterables.transform(contents, new Function<WindowedValue<T>, T>() { @Override @@ -104,6 +112,11 @@ public final class PCollectionViewTesting { */ public static class LengthViewFn<T> extends ViewFn<Iterable<WindowedValue<T>>, Long> { @Override + public Materialization<Iterable<WindowedValue<T>>> getMaterialization() { + return Materializations.iterable(); + } + + @Override public Long apply(Iterable<WindowedValue<T>> contents) { return (long) Iterables.size(contents); } @@ -121,6 +134,11 @@ public final class PCollectionViewTesting { } @Override + public Materialization<Iterable<WindowedValue<ElemT>>> getMaterialization() { + return Materializations.iterable(); + } + + @Override public ViewT apply(Iterable<WindowedValue<ElemT>> contents) { return value; }