Hello,

I am trying to run SparkGraphComputer on a JanusGraph backed by Cassandra and ElasticSearch. I have previously verified that I am able to run SparkGraphComputer on a local Spark standalone cluster.

I am now trying to run it on YARN. I have a local YARN cluster running and I have verified that it can run Spark jobs.

I followed the following links:

http://yaaics.blogspot.com/2017/07/configuring-janusgraph-for-spark-yarn.html

http://tinkerpop.apache.org/docs/3.4.6/recipes/#olap-spark-yarn

And here is my read-cql-yarn.properties file:

gremlin.graph=org.apache.tinkerpop.gremlin.hadoop.structure.HadoopGraph

gremlin.hadoop.graphReader=org.janusgraph.hadoop.formats.cql.CqlInputFormat

gremlin.hadoop.graphWriter=org.apache.hadoop.mapreduce.lib.output.NullOutputFormat

gremlin.hadoop.jarsInDistributedCache=true

gremlin.hadoop.inputLocation=none

gremlin.hadoop.outputLocation=output

gremlin.spark.persistContext=true

#

# JanusGraph Cassandra InputFormat configuration

#

# These properties defines the connection properties which were used while write data to JanusGraph.

janusgraphmr.ioformat.conf.storage.backend=cql

# This specifies the hostname & port for Cassandra data store.

janusgraphmr.ioformat.conf.storage.hostname=127.0.0.1

janusgraphmr.ioformat.conf.storage.port=9042

# This specifies the keyspace where data is stored.

janusgraphmr.ioformat.conf.storage.cql.keyspace=janusgraph

# This defines the indexing backend configuration used while writing data to JanusGraph.

janusgraphmr.ioformat.conf.index.search.backend=elasticsearch

janusgraphmr.ioformat.conf.index.search.hostname=127.0.0.1

# Use the appropriate properties for the backend when using a different storage backend (HBase) or indexing backend (Solr).

#

# Apache Cassandra InputFormat configuration

#

cassandra.input.partitioner.class=org.apache.cassandra.dht.Murmur3Partitioner

cassandra.input.widerows=true

#

# SparkGraphComputer Configuration

#

spark.master=yarn

spark.submit.deployMode=client

spark.executor.memory=1g

spark.yarn.dist.archives=/tmp/spark-gremlin.zip

spark.yarn.dist.files=/Users/my_comp/Downloads/janusgraph-0.5.2/lib/janusgraph-cql-0.5.2.jar

spark.yarn.appMasterEnv.CLASSPATH=/Users/my_comp/Downloads/hadoop-2.7.2/etc/hadoop:./spark-gremlin.zip/*

spark.executor.extraClassPath=/Users/my_comp/Downloads/hadoop-2.7.2/etc/hadoop:/Users/my_comp/Downloads/janusgraph-0.5.2/lib/janusgraph-cql-0.5.2.jar:./spark-gremlin.zip/*

spark.driver.extraLibraryPath=/Users/my_comp/Downloads/hadoop-2.7.2/lib/native:/Users/my_comp/Downloads/hadoop-2.7.2/lib/native/Linux-amd64-64

spark.executor.extraLibraryPath=/Users/my_comp/Downloads/hadoop-2.7.2/lib/native:/Users/my_comp/Downloads/hadoop-2.7.2/lib/native/Linux-amd64-64

spark.serializer=org.apache.spark.serializer.KryoSerializer

spark.kryo.registrator=org.janusgraph.hadoop.serialize.JanusGraphKryoRegistrator

After a bunch of trial and error, I was able to get it to a point where I see containers starting up on my YARN Resource manager UI (port 8088)

Here is the code I am running (it's a simple count):

gremlin> graph = GraphFactory.open('conf/hadoop-graph/read-cql-yarn.properties')

==>hadoopgraph[cqlinputformat->nulloutputformat]

gremlin> g = graph.traversal().withComputer(SparkGraphComputer)

==>graphtraversalsource[hadoopgraph[cqlinputformat->nulloutputformat], sparkgraphcomputer]

gremlin> g.V().count()

However I am encountering the following failure:

18:49:03 ERROR org.apache.spark.scheduler.TaskSetManager - Task 2 in stage 0.0 failed 4 times; aborting job

org.apache.spark.SparkException: Job aborted due to stage failure: Task 2 in stage 0.0 failed 4 times, most recent failure: Lost task 2.3 in stage 0.0 (TID 10, 192.168.1.160, executor 1): java.lang.IllegalStateException: unread block data

at java.io.ObjectInputStream$BlockDataInputStream.setBlockDataMode(ObjectInputStream.java:2862)

at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1682)

at java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2366)

at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:2290)

at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2148)

at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1647)

at java.io.ObjectInputStream.readObject(ObjectInputStream.java:483)

at java.io.ObjectInputStream.readObject(ObjectInputStream.java:441)

at org.apache.spark.serializer.JavaDeserializationStream.readObject(JavaSerializer.scala:75)

at org.apache.spark.serializer.JavaSerializerInstance.deserialize(JavaSerializer.scala:114)

at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:370)

at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)

at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)

at java.lang.Thread.run(Thread.java:748)

Would really appricate it if someone could shed some light on this error and advise on next steps!

Thank you!