You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by am...@apache.org on 2016/03/10 21:58:52 UTC
[27/50] [abbrv] incubator-beam git commit: Add NullWritableCoder and
test.
Add NullWritableCoder and test.
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/8762b267
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/8762b267
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/8762b267
Branch: refs/heads/master
Commit: 8762b26717e981d02a44e2df5ddd3ef08f10e2f9
Parents: 7838865
Author: Tom White <to...@cloudera.com>
Authored: Thu Aug 20 12:54:38 2015 +0100
Committer: Tom White <to...@cloudera.com>
Committed: Thu Mar 10 11:15:16 2016 +0000
----------------------------------------------------------------------
.../dataflow/hadoop/NullWritableCoder.java | 71 ++++++++++++++++++++
.../cloudera/dataflow/hadoop/WritableCoder.java | 4 ++
.../dataflow/hadoop/WritableCoderTest.java | 42 ++++++++++++
3 files changed, 117 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/8762b267/runners/spark/src/main/java/com/cloudera/dataflow/hadoop/NullWritableCoder.java
----------------------------------------------------------------------
diff --git a/runners/spark/src/main/java/com/cloudera/dataflow/hadoop/NullWritableCoder.java b/runners/spark/src/main/java/com/cloudera/dataflow/hadoop/NullWritableCoder.java
new file mode 100644
index 0000000..5e5d391
--- /dev/null
+++ b/runners/spark/src/main/java/com/cloudera/dataflow/hadoop/NullWritableCoder.java
@@ -0,0 +1,71 @@
+/*
+ * Copyright (c) 2015, Cloudera, Inc. All Rights Reserved.
+ *
+ * Cloudera, Inc. 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
+ *
+ * This software 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 com.cloudera.dataflow.hadoop;
+
+import java.io.InputStream;
+import java.io.OutputStream;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.google.cloud.dataflow.sdk.coders.Coder;
+import org.apache.hadoop.io.NullWritable;
+
+public final class NullWritableCoder extends WritableCoder<NullWritable> {
+ private static final long serialVersionUID = 1L;
+
+ @JsonCreator
+ public static NullWritableCoder of() {
+ return INSTANCE;
+ }
+
+ private static final NullWritableCoder INSTANCE = new NullWritableCoder();
+
+ private NullWritableCoder() {
+ super(NullWritable.class);
+ }
+
+ @Override
+ public void encode(NullWritable value, OutputStream outStream, Context context) {
+ // nothing to write
+ }
+
+ @Override
+ public NullWritable decode(InputStream inStream, Context context) {
+ return NullWritable.get();
+ }
+
+ @Override
+ public boolean consistentWithEquals() {
+ return true;
+ }
+
+ /**
+ * Returns true since registerByteSizeObserver() runs in constant time.
+ */
+ @Override
+ public boolean isRegisterByteSizeObserverCheap(NullWritable value, Context context) {
+ return true;
+ }
+
+ @Override
+ protected long getEncodedElementByteSize(NullWritable value, Context context) {
+ return 0;
+ }
+
+ @Override
+ public void verifyDeterministic() throws Coder.NonDeterministicException {
+ // NullWritableCoder is deterministic
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/8762b267/runners/spark/src/main/java/com/cloudera/dataflow/hadoop/WritableCoder.java
----------------------------------------------------------------------
diff --git a/runners/spark/src/main/java/com/cloudera/dataflow/hadoop/WritableCoder.java b/runners/spark/src/main/java/com/cloudera/dataflow/hadoop/WritableCoder.java
index dbc6779..ea47109 100644
--- a/runners/spark/src/main/java/com/cloudera/dataflow/hadoop/WritableCoder.java
+++ b/runners/spark/src/main/java/com/cloudera/dataflow/hadoop/WritableCoder.java
@@ -28,6 +28,7 @@ import com.google.cloud.dataflow.sdk.coders.Coder;
import com.google.cloud.dataflow.sdk.coders.CoderException;
import com.google.cloud.dataflow.sdk.coders.StandardCoder;
import com.google.cloud.dataflow.sdk.util.CloudObject;
+import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Writable;
/**
@@ -54,6 +55,9 @@ public class WritableCoder<T extends Writable> extends StandardCoder<T> {
* @return a {@code WritableCoder} instance for the provided element class
*/
public static <T extends Writable> WritableCoder<T> of(Class<T> clazz) {
+ if (clazz.equals(NullWritable.class)) {
+ return (WritableCoder<T>) NullWritableCoder.of();
+ }
return new WritableCoder<>(clazz);
}
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/8762b267/runners/spark/src/test/java/com/cloudera/dataflow/hadoop/WritableCoderTest.java
----------------------------------------------------------------------
diff --git a/runners/spark/src/test/java/com/cloudera/dataflow/hadoop/WritableCoderTest.java b/runners/spark/src/test/java/com/cloudera/dataflow/hadoop/WritableCoderTest.java
new file mode 100644
index 0000000..29a73b6
--- /dev/null
+++ b/runners/spark/src/test/java/com/cloudera/dataflow/hadoop/WritableCoderTest.java
@@ -0,0 +1,42 @@
+/*
+ * Copyright (c) 2015, Cloudera, Inc. All Rights Reserved.
+ *
+ * Cloudera, Inc. 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
+ *
+ * This software 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 com.cloudera.dataflow.hadoop;
+
+import com.google.cloud.dataflow.sdk.testing.CoderProperties;
+import org.apache.hadoop.io.IntWritable;
+import org.apache.hadoop.io.NullWritable;
+import org.junit.Test;
+
+/**
+ * Tests for WritableCoder.
+ */
+public class WritableCoderTest {
+
+ @Test
+ public void testIntWritableEncoding() throws Exception {
+ IntWritable value = new IntWritable(42);
+ WritableCoder<IntWritable> coder = WritableCoder.of(IntWritable.class);
+
+ CoderProperties.coderDecodeEncodeEqual(coder, value);
+ }
+
+ @Test
+ public void testNullWritableEncoding() throws Exception {
+ WritableCoder<NullWritable> coder = WritableCoder.of(NullWritable.class);
+
+ CoderProperties.coderDecodeEncodeEqual(coder, NullWritable.get());
+ }
+}