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/04/06 01:09:38 UTC
[2/2] beam git commit: Add Initial Materializations
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;
}