Running OLAP queries in Janusgraph with the help of Spark

[jthobhani]

Hello Folks, have been trying from last few days to run OLAP queries on a huge graph of around 4TB keyspace size consisting of millions of vertices & edges and billions of partitions but not being successful. So was researching a bit by going over the Core Storage in cassandra tables like edgestore and janusgraph_id table where the data is stored in encrypted and serialized format. Also saw an online meetup where they said that g.V().id() may go through to get the count but when I ran through spark looks like it also needs a good amount of memory as it's throwing OOM's.

Is it possible to decrypt and deserialize the data from edgestore table directly? Wanted to achieve various functionalities like data deletion on a periodic basis, run analytical queries to get the trending data from graph etc. Would love to get some ideas and relevant answers on it.

[li-boxuan]

run OLAP queries on a huge graph of around 4TB keyspace size consisting of millions of vertices & edges and billions of partitions but not being successful.

Can you elaborate on that? What queries did you try, and what was your environment setup?

Asking because "OLAP" is a vague term. Not sure what you actually referred to here.

[jthobhani]

Sure @li-boxuan, below are the few queries which we tried. In terms of environment setup we have a spark Hadoop cluster and this are the adhoc jobs which we are trying to run:-

1. Firstly, we tried to just fetch the labels g.V().label().toList() so that it doesn't iterate over all the properties of vertices but there also it's failing due to OOM's then we raised the yarn scheduler memory to 100G but then cassandra nodes started getting timed out.
2. We went through online meetup of JG where they have discussed regarding running the analytical queries and they suggested to iterate over the id's instead, so we tried g.V().id().toStream().count() but it also failed throwing OOM's.

Please lemme know if you need any more detailed information.

[li-boxuan]

I assume you are using g = graph.traversal().withComputer(SparkGraphComputer), followed by the traversals. It leverages a special "VertexProgram" called "TraversalVertexProgram": https://tinkerpop.apache.org/docs/current/reference/#traversalvertexprogram, which in my opinion is not fully optimized.

For your first query, could you try doing this instead?

result = graph.compute(SparkGraphComputer.class)
    .vertexProperties(__.properties("dummy"))
    .program(TraversalVertexProgram.build().traversal(g.V().label().toList()).create()).submit().get()
result.memory().asMap()

and observe if there's significant savings in memory. If so, then it's a flaw of TraversalVertexProgram, and your OOM can be explained. Otherwise, we may need further explorations.

Are all your use cases simple as that? If so, you could use Spark to load all vertices like this:

        JavaPairRDD<Object, VertexWritable> verticesRDD = SparkIOUtil.loadVertices(sparkGraphConfiguration, sparkContext);
        verticesRDD.values().foreachPartition(partition -> {
            // SparkRDDs are immutable, so we have to use normal JanusGraph transactions to
            // write updated data back to JanusGraph. For each partition, we open a transaction.
            JanusGraph graph;
            try {
                graph = JanusGraphFactory.open(resource.toURI().getPath());
            } catch (Exception ex) {
                System.out.println(ex);
                return;
            }

            JanusGraph finalGraph = graph;
            partition.forEachRemaining(
                vertexWritable -> {

A complete example is available here: https://github.com/Citegraph/citegraph/blob/main/backend/src/main/java/io/citegraph/data/spark/loader/AuthorRefEdgeLoader.java#L45

This will give you more fine-grained control, and will use less memory than TraversalVertexProgram. If you still get OOM, you should tune gremlin.spark.graphStorageLevel to something like DISK_ONLY. If you still get OOM, then final resolution would be to use MapReduce.

[jthobhani]

Hello @li-boxuan tried the SparkIOUtil loadVertices way but getting the below error, tried many different things but not able to figure out. Can you please take a look at it and lemme know if you also faced the same and were able to resolve it?

Using BaseConfiguration :
Caused by: java.io.NotSerializableException: org.apache.commons.configuration2.BaseConfiguration
Serialization stack:

• object not serializable (class: org.apache.commons.configuration2.BaseConfiguration, value: org.apache.commons.configuration2.BaseConfiguration@1f10a8d)
• element of array (index: 0)
• array (class [Ljava.lang.Object;, size 1)
• field (class: java.lang.invoke.SerializedLambda, name: capturedArgs, type: class [Ljava.lang.Object;)
• object (class java.lang.invoke.SerializedLambda, SerializedLambda[capturingClass=class com.qualys.ctdb.spark.jobs.JGVertexIdJob, functionalInterfaceMethod=org/apache/spark/api/java/function/VoidFunction.call:(Ljava/lang/Object;)V, implementation=invokeStatic com/qualys/ctdb/spark/jobs/JGVertexIdJob.lambda$main$9739ec9f$1:(Lorg/apache/commons/configuration2/Configuration;Ljava/util/Iterator;)V, instantiatedMethodType=(Ljava/util/Iterator;)V, numCaptured=1])
• writeReplace data (class: java.lang.invoke.SerializedLambda)
• object (class com.qualys.ctdb.spark.jobs.JGVertexIdJob$$Lambda$lambda$main$9739ec9f$1$2962002098/0x0000000000002eee, com.qualys.ctdb.spark.jobs.JGVertexIdJob$$Lambda$lambda$main$9739ec9f$1$2962002098/0x0000000000002eee@15b36503)
• element of array (index: 0)
• array (class [Ljava.lang.Object;, size 1)
• field (class: java.lang.invoke.SerializedLambda, name: capturedArgs, type: class [Ljava.lang.Object;)
• object (class java.lang.invoke.SerializedLambda, SerializedLambda[capturingClass=interface org.apache.spark.api.java.JavaRDDLike, functionalInterfaceMethod=scala/Function1.apply:(Ljava/lang/Object;)Ljava/lang/Object;, implementation=invokeStatic org/apache/spark/api/java/JavaRDDLike.$anonfun$foreachPartition$1$adapted:(Lorg/apache/spark/api/java/function/VoidFunction;Lscala/collection/Iterator;)Ljava/lang/Object;, instantiatedMethodType=(Lscala/collection/Iterator;)Ljava/lang/Object;, numCaptured=1])
• writeReplace data (class: java.lang.invoke.SerializedLambda)
• object (class org.apache.spark.api.java.JavaRDDLike$$Lambda$$anonfun$foreachPartition$1$adapted$816664089/0x0000000000002eef, org.apache.spark.api.java.JavaRDDLike$$Lambda$$anonfun$foreachPartition$1$adapted$816664089/0x0000000000002eef@15b316fa)
  at org.apache.spark.serializer.SerializationDebugger$.improveException(SerializationDebugger.scala:41)
  at org.apache.spark.serializer.JavaSerializationStream.writeObject(JavaSerializer.scala:49)
  at org.apache.spark.serializer.JavaSerializerInstance.serialize(JavaSerializer.scala:115)
  at org.apache.spark.util.ClosureCleaner$.ensureSerializable(ClosureCleaner.scala:441)
  ... 17 more

Using MapConfiguration:
Caused by: java.io.NotSerializableException: org.apache.commons.configuration2.MapConfiguration
Serialization stack:

• object not serializable (class: org.apache.commons.configuration2.MapConfiguration, value: MapConfiguration [map={janusgraphmr.ioformat.conf.index.ctdbgraphsearch.index-name=opensearchgraph01, janusgraphmr.ioformat.conf.storage.hostname=10.44.206.21,10.44.206.22,10.44.206.23,10.44.206.24,10.44.206.25, gremlin.graph=org.apache.tinkerpop.gremlin.hadoop.structure.HadoopGraph, janusgraphmr.ioformat.conf.index.ctdbgraphsearch.port=50140, janusgraphmr.ioformat.conf.index.ctdbgraphsearch.elasticsearch.ssl.truststore.password=changeit, janusgraphmr.ioformat.conf.storage.cql.local-datacenter=dc2, gremlin.hadoop.graphWriter=org.apache.hadoop.mapreduce.lib.output.NullOutputFormat, gremlin.hadoop.jarsInDistributedCache=true, cassandra.input.partitioner.class=org.apache.cassandra.dht.Murmur3Partitioner, janusgraphmr.ioformat.conf.index.ctdbgraphsearch.hostname=qosdatacmn01.mp04.eng.sjc01.qualys.com:50140,qosdatacmn02.mp04.eng.sjc01.qualys.com:50140,qosdatacmn03.mp04.eng.sjc01.qualys.com:50140, janusgraphmr.ioformat.conf.index.ctdbgraphsearch.elasticsearch.http.auth.type=basic, janusgraphmr.ioformat.conf.index.ctdbgraphsearch.backend=elasticsearch, janusgraphmr.ioformat.conf.index.ctdbgraphsearch.elasticsearch.ssl.enabled=true, janusgraphmr.ioformat.conf.metrics.enabled=false, gremlin.hadoop.outputLocation=output, gremlin.spark.persistContext=true, cassandra.input.widerows=true, janusgraphmr.ioformat.conf.storage.port=50126, janusgraphmr.ioformat.conf.storage.backend=cql, janusgraphmr.ioformat.conf.index.ctdbgraphsearch.elasticsearch.ssl.truststore.location=/etc/pki/ca-trust/extracted/java/cacerts, janusgraphmr.ioformat.conf.storage.cql.keyspace=ctdbgraph, gremlin.hadoop.graphReader=org.janusgraph.hadoop.formats.cql.CqlInputFormat, gremlin.hadoop.inputLocation=none}, trimmingDisabled=false])
• element of array (index: 0)
• array (class [Ljava.lang.Object;, size 1)
• field (class: java.lang.invoke.SerializedLambda, name: capturedArgs, type: class [Ljava.lang.Object;)
• object (class java.lang.invoke.SerializedLambda, SerializedLambda[capturingClass=class com.qualys.ctdb.spark.jobs.JGVertexIdJob, functionalInterfaceMethod=org/apache/spark/api/java/function/VoidFunction.call:(Ljava/lang/Object;)V, implementation=invokeStatic com/qualys/ctdb/spark/jobs/JGVertexIdJob.lambda$main$9739ec9f$1:(Lorg/apache/commons/configuration2/Configuration;Ljava/util/Iterator;)V, instantiatedMethodType=(Ljava/util/Iterator;)V, numCaptured=1])
• writeReplace data (class: java.lang.invoke.SerializedLambda)
• object (class com.qualys.ctdb.spark.jobs.JGVertexIdJob$$Lambda$lambda$main$9739ec9f$1$2962002098/0x0000000000002eed, com.qualys.ctdb.spark.jobs.JGVertexIdJob$$Lambda$lambda$main$9739ec9f$1$2962002098/0x0000000000002eed@21e65412)
• element of array (index: 0)
• array (class [Ljava.lang.Object;, size 1)
• field (class: java.lang.invoke.SerializedLambda, name: capturedArgs, type: class [Ljava.lang.Object;)
• object (class java.lang.invoke.SerializedLambda, SerializedLambda[capturingClass=interface org.apache.spark.api.java.JavaRDDLike, functionalInterfaceMethod=scala/Function1.apply:(Ljava/lang/Object;)Ljava/lang/Object;, implementation=invokeStatic org/apache/spark/api/java/JavaRDDLike.$anonfun$foreachPartition$1$adapted:(Lorg/apache/spark/api/java/function/VoidFunction;Lscala/collection/Iterator;)Ljava/lang/Object;, instantiatedMethodType=(Lscala/collection/Iterator;)Ljava/lang/Object;, numCaptured=1])
• writeReplace data (class: java.lang.invoke.SerializedLambda)
• object (class org.apache.spark.api.java.JavaRDDLike$$Lambda$$anonfun$foreachPartition$1$adapted$816664089/0x0000000000002eee, org.apache.spark.api.java.JavaRDDLike$$Lambda$$anonfun$foreachPartition$1$adapted$816664089/0x0000000000002eee@21e6a59b)
  at org.apache.spark.serializer.SerializationDebugger$.improveException(SerializationDebugger.scala:41)
  at org.apache.spark.serializer.JavaSerializationStream.writeObject(JavaSerializer.scala:49)
  at org.apache.spark.serializer.JavaSerializerInstance.serialize(JavaSerializer.scala:115)
  at org.apache.spark.util.ClosureCleaner$.ensureSerializable(ClosureCleaner.scala:441)

[li-boxuan]

From your opening comment it seems you were able to run some spark-backed OLAP queries. Did you use a different setup or different JanusGraph version this time? Did you change any config? IIRC there might be package conflicts between some packages that JanusGraph depends on and what Spark depends on, and a manual conflict resolution would be needed. From your error log, it is unclear whether it's a dependency conflict problem, though. It seems more likely you have some issues in your configs.

[jthobhani]

Yes, we were. At that time we were leveraging JG 0.6.2 version but now with this we are leveraging 1.0.0 which has SparkIOUtil class and have also updated the JG server on our side.

Here are the serializer configs in our server :-
serializers:

• { className: org.apache.tinkerpop.gremlin.util.ser.GraphBinaryMessageSerializerV1, config: { ioRegistries: [org.janusgraph.graphdb.tinkerpop.JanusGraphIoRegistry] }}
• { className: org.apache.tinkerpop.gremlin.util.ser.GraphBinaryMessageSerializerV1, config: { serializeResultToString: true }}
• { className: org.apache.tinkerpop.gremlin.util.ser.GraphSONMessageSerializerV3, config: { ioRegistries: [org.janusgraph.graphdb.tinkerpop.JanusGraphIoRegistry] }}

Older serialization versions for backwards compatibility:

• { className: org.apache.tinkerpop.gremlin.util.ser.GraphSONMessageSerializerV2, config: { ioRegistries: [org.janusgraph.graphdb.tinkerpop.JanusGraphIoRegistry] }}
• { className: org.apache.tinkerpop.gremlin.util.ser.GraphSONMessageSerializerV1, config: { ioRegistries: [org.janusgraph.graphdb.tinkerpop.JanusGraphIoRegistryV1d0] }}
• { className: org.apache.tinkerpop.gremlin.util.ser.GraphSONUntypedMessageSerializerV1, config: { ioRegistries: [org.janusgraph.graphdb.tinkerpop.JanusGraphIoRegistryV1d0] }}

[jthobhani]

Hello @li-boxuan I fixed all the above issues and triggered a spark job to get the groupCount by iterating over 10k vertices in a single loop. Getting the below error now related to unread blocks, Can you suggest if you have also faced before or have any solution in mind?

Caused by: java.lang.IllegalStateException: unread block data
at java.io.ObjectInputStream$BlockDataInputStream.setBlockDataMode(ObjectInputStream.java:2934)
at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1704)
at java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2431)
at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:2355)
at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2213)
at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1669)
at java.io.ObjectInputStream.readObject(ObjectInputStream.java:503)
at java.io.ObjectInputStream.readObject(ObjectInputStream.java:461)
at org.apache.spark.scheduler.DirectTaskResult.$anonfun$readExternal$2(TaskResult.scala:77)
at org.apache.spark.scheduler.DirectTaskResult.$anonfun$readExternal$2$adapted(TaskResult.scala:76)
at scala.collection.immutable.Range.foreach(Range.scala:158)
at org.apache.spark.scheduler.DirectTaskResult.$anonfun$readExternal$1(TaskResult.scala:76)
at scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)
at org.apache.spark.util.Utils$.tryOrIOException(Utils.scala:1495)

[li-boxuan]

If you ask on the user mailing list, maybe hadoopmarc would be able to help you. He is an expert of janusgraph - spark interactions.

[li-boxuan]

I didn't see this before.

[hadoopmarc]

Caused by: java.lang.IllegalStateException: unread block data

This exception also has to do with the graph size and size of the keyspace partitions. It may even be related to the memory configs of the spark driver.
Anyway, there is a fundamental reason why this cannot work for anything more complex than a count. For OLAP with spark to work the entire graph has to fit in the cluster memory as java objects, so think 40 TB. Otherwise, the graph has to be written to and read from disk for every traversal step. So you have to think about ways to reduce data size first. Some mechanisms that might be exploited:

• https://tinkerpop.apache.org/docs/current/reference/#graph-filter
• https://docs.janusgraph.org/advanced-topics/partitioning/