You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tinkerpop.apache.org by ok...@apache.org on 2015/09/24 02:51:31 UTC
incubator-tinkerpop git commit: different errors in master/. I think
we are closer on 3.1.0 than with 3.0.2.
Repository: incubator-tinkerpop
Updated Branches:
refs/heads/spark-gryo-tp31 [created] 9110a1e70
different errors in master/. I think we are closer on 3.1.0 than with 3.0.2.
Project: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/commit/9110a1e7
Tree: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/tree/9110a1e7
Diff: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/diff/9110a1e7
Branch: refs/heads/spark-gryo-tp31
Commit: 9110a1e707d54da903ecb21d073d2d877e864e77
Parents: c6de3bf
Author: Marko A. Rodriguez <ok...@gmail.com>
Authored: Wed Sep 23 18:51:27 2015 -0600
Committer: Marko A. Rodriguez <ok...@gmail.com>
Committed: Wed Sep 23 18:51:27 2015 -0600
----------------------------------------------------------------------
.../io/gryo/GryoDeserializationStream.java | 55 ++++++++++++++
.../io/gryo/GryoSerializationStream.java | 70 ++++++++++++++++++
.../computer/io/gryo/GryoSerializer.java | 62 ++++++++++++++++
.../io/gryo/GryoSerializerInstance.java | 76 ++++++++++++++++++++
.../computer/HadoopSparkGraphProvider.java | 3 +-
5 files changed, 265 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/9110a1e7/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/process/computer/io/gryo/GryoDeserializationStream.java
----------------------------------------------------------------------
diff --git a/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/process/computer/io/gryo/GryoDeserializationStream.java b/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/process/computer/io/gryo/GryoDeserializationStream.java
new file mode 100644
index 0000000..f5e36f7
--- /dev/null
+++ b/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/process/computer/io/gryo/GryoDeserializationStream.java
@@ -0,0 +1,55 @@
+/*
+ * 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.tinkerpop.gremlin.spark.process.computer.io.gryo;
+
+import org.apache.spark.serializer.DeserializationStream;
+import org.apache.tinkerpop.shaded.kryo.io.Input;
+import scala.reflect.ClassTag;
+
+import java.io.IOException;
+import java.io.InputStream;
+
+/**
+ * @author Marko A. Rodriguez (http://markorodriguez.com)
+ */
+public class GryoDeserializationStream extends DeserializationStream {
+
+ private final InputStream inputStream;
+ private final GryoSerializerInstance serializer;
+
+ public GryoDeserializationStream(final GryoSerializerInstance serializer, final InputStream inputStream) {
+ this.serializer = serializer;
+ this.inputStream = inputStream;
+ }
+
+ @Override
+ public <T> T readObject(final ClassTag<T> classTag) {
+ return (T) this.serializer.getKryo().readClassAndObject(new Input(this.inputStream));
+ }
+
+ @Override
+ public void close() {
+ try {
+ this.inputStream.close();
+ } catch (final IOException e) {
+ throw new IllegalStateException(e.getMessage(), e);
+ }
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/9110a1e7/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/process/computer/io/gryo/GryoSerializationStream.java
----------------------------------------------------------------------
diff --git a/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/process/computer/io/gryo/GryoSerializationStream.java b/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/process/computer/io/gryo/GryoSerializationStream.java
new file mode 100644
index 0000000..e5981a0
--- /dev/null
+++ b/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/process/computer/io/gryo/GryoSerializationStream.java
@@ -0,0 +1,70 @@
+/*
+ * 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.tinkerpop.gremlin.spark.process.computer.io.gryo;
+
+import org.apache.spark.serializer.SerializationStream;
+import org.apache.spark.serializer.SerializerInstance;
+import scala.reflect.ClassTag;
+
+import java.io.IOException;
+import java.io.OutputStream;
+
+/**
+ * @author Marko A. Rodriguez (http://markorodriguez.com)
+ */
+public class GryoSerializationStream extends SerializationStream {
+
+ private final OutputStream outputStream;
+ private final SerializerInstance serializer;
+
+ public GryoSerializationStream(final GryoSerializerInstance serializer, final OutputStream outputStream) {
+ this.outputStream = outputStream;
+ this.serializer = serializer;
+ }
+
+ @Override
+ public <T> SerializationStream writeObject(final T t, final ClassTag<T> classTag) {
+ try {
+ this.outputStream.write(this.serializer.serialize(t, classTag).array());
+ this.outputStream.flush();
+ } catch (final IOException e) {
+ throw new IllegalStateException(e.getMessage(), e);
+ }
+ return this;
+ }
+
+ @Override
+ public void flush() {
+ try {
+ this.outputStream.flush();
+ } catch (final IOException e) {
+ throw new IllegalStateException(e.getMessage(), e);
+ }
+ }
+
+ @Override
+ public void close() {
+ try {
+ this.outputStream.close();
+ } catch (final IOException e) {
+ throw new IllegalStateException(e.getMessage(), e);
+ }
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/9110a1e7/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/process/computer/io/gryo/GryoSerializer.java
----------------------------------------------------------------------
diff --git a/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/process/computer/io/gryo/GryoSerializer.java b/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/process/computer/io/gryo/GryoSerializer.java
new file mode 100644
index 0000000..c7c1468
--- /dev/null
+++ b/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/process/computer/io/gryo/GryoSerializer.java
@@ -0,0 +1,62 @@
+/*
+ * 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.tinkerpop.gremlin.spark.process.computer.io.gryo;
+
+import org.apache.spark.SerializableWritable;
+import org.apache.spark.api.python.PythonBroadcast;
+import org.apache.spark.broadcast.HttpBroadcast;
+import org.apache.spark.scheduler.CompressedMapStatus;
+import org.apache.spark.serializer.Serializer;
+import org.apache.spark.serializer.SerializerInstance;
+import org.apache.spark.util.SerializableConfiguration;
+import org.apache.tinkerpop.gremlin.hadoop.structure.io.ObjectWritable;
+import org.apache.tinkerpop.gremlin.hadoop.structure.io.VertexWritable;
+import org.apache.tinkerpop.gremlin.spark.process.computer.payload.MessagePayload;
+import org.apache.tinkerpop.gremlin.spark.process.computer.payload.ViewIncomingPayload;
+import org.apache.tinkerpop.gremlin.spark.process.computer.payload.ViewOutgoingPayload;
+import org.apache.tinkerpop.gremlin.spark.process.computer.payload.ViewPayload;
+import org.apache.tinkerpop.gremlin.structure.io.gryo.GryoMapper;
+import org.apache.tinkerpop.shaded.kryo.serializers.JavaSerializer;
+import scala.Tuple2;
+
+/**
+ * @author Marko A. Rodriguez (http://markorodriguez.com)
+ */
+public class GryoSerializer extends Serializer {
+ @Override
+ public SerializerInstance newInstance() {
+ return new GryoSerializerInstance(
+ GryoMapper.build().
+ addCustom(SerializableWritable.class, new JavaSerializer()).
+ addCustom(Tuple2.class, new JavaSerializer()).
+ addCustom(CompressedMapStatus.class, new JavaSerializer()).
+ addCustom(HttpBroadcast.class, new JavaSerializer()).
+ addCustom(PythonBroadcast.class, new JavaSerializer()).
+ addCustom(MessagePayload.class, new JavaSerializer()).
+ addCustom(ViewIncomingPayload.class, new JavaSerializer()).
+ addCustom(ViewOutgoingPayload.class, new JavaSerializer()).
+ addCustom(ViewPayload.class, new JavaSerializer()).
+ addCustom(SerializableConfiguration.class, new JavaSerializer()).
+ addCustom(VertexWritable.class, new JavaSerializer()).
+ addCustom(ObjectWritable.class, new JavaSerializer()).
+ create().createMapper());
+ // kryo.register(org.apache.spark.serializer.JavaIterableWrapperSerializer..MODULE$.wrapperClass(), new JavaIterableWrapperSerializer());
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/9110a1e7/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/process/computer/io/gryo/GryoSerializerInstance.java
----------------------------------------------------------------------
diff --git a/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/process/computer/io/gryo/GryoSerializerInstance.java b/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/process/computer/io/gryo/GryoSerializerInstance.java
new file mode 100644
index 0000000..212abe2
--- /dev/null
+++ b/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/process/computer/io/gryo/GryoSerializerInstance.java
@@ -0,0 +1,76 @@
+/*
+ * 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.tinkerpop.gremlin.spark.process.computer.io.gryo;
+
+import org.apache.spark.serializer.DeserializationStream;
+import org.apache.spark.serializer.SerializationStream;
+import org.apache.spark.serializer.SerializerInstance;
+import org.apache.tinkerpop.shaded.kryo.Kryo;
+import org.apache.tinkerpop.shaded.kryo.io.Input;
+import org.apache.tinkerpop.shaded.kryo.io.Output;
+import scala.reflect.ClassTag;
+
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.nio.ByteBuffer;
+
+/**
+ * @author Marko A. Rodriguez (http://markorodriguez.com)
+ */
+public class GryoSerializerInstance extends SerializerInstance {
+
+ private final Kryo kryo;
+
+ public GryoSerializerInstance(final Kryo kryo) {
+ this.kryo = kryo;
+ }
+
+ @Override
+ public <T> ByteBuffer serialize(final T t, final ClassTag<T> classTag) {
+ final Output output = new Output(100000);
+ this.kryo.writeClassAndObject(output, t);
+ return ByteBuffer.wrap(output.getBuffer());
+ }
+
+ @Override
+ public <T> T deserialize(final ByteBuffer byteBuffer, final ClassTag<T> classTag) {
+ return (T) this.kryo.readClassAndObject(new Input(byteBuffer.array()));
+ }
+
+ @Override
+ public <T> T deserialize(final ByteBuffer byteBuffer, final ClassLoader classLoader, final ClassTag<T> classTag) {
+ this.kryo.setClassLoader(classLoader);
+ return (T) this.kryo.readClassAndObject(new Input(byteBuffer.array()));
+ }
+
+ @Override
+ public SerializationStream serializeStream(final OutputStream outputStream) {
+ return new GryoSerializationStream(this, outputStream);
+ }
+
+ @Override
+ public DeserializationStream deserializeStream(final InputStream inputStream) {
+ return new GryoDeserializationStream(this, inputStream);
+ }
+
+ public Kryo getKryo() {
+ return this.kryo;
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/9110a1e7/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/process/computer/HadoopSparkGraphProvider.java
----------------------------------------------------------------------
diff --git a/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/process/computer/HadoopSparkGraphProvider.java b/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/process/computer/HadoopSparkGraphProvider.java
index c7642c8..7ed741b 100644
--- a/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/process/computer/HadoopSparkGraphProvider.java
+++ b/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/process/computer/HadoopSparkGraphProvider.java
@@ -114,7 +114,8 @@ public final class HadoopSparkGraphProvider extends AbstractGraphProvider {
put("mapreduce.job.reduces", 4);
/// spark configuration
put("spark.master", "local[4]");
- put("spark.serializer", "org.apache.spark.serializer.KryoSerializer");
+ // put("spark.serializer", "org.apache.spark.serializer.KryoSerializer");
+ put("spark.serializer","org.apache.tinkerpop.gremlin.spark.process.computer.io.gryo.GryoSerializer");
// put("spark.kryo.registrationRequired",true);
}};
}