JVM Errors/Java Nullpointer exceptions

[PallaviPersistent]

Branch, version, commit

OS and Environment
CentOS Linux 7.3

JVM version
JDK 1.8

Scala version
Scala version 2.11

Kafka and Cassandra versions and setup
Cassandra version 3.2

Spark version if used
Spark 2.0

Deployed mode
(client/cluster on Spark Standalone/YARN/Mesos/EMR or default)
Standalone

Actual (wrong) behavior
scala> dfResult.write.format("filodb.spark").option("dataset", "flow_raw").option("row_keys", "exportMs").option("partition_keys", "protocolId").mode(SaveMode.Append).save()
[INFO] [05/10/2018 23:15:15.943] [main] [StatsDExtension(akka://kamon)] Starting the Kamon(StatsD) extension
[Stage 8:====> (1 + 11) / 12]#

A fatal error has been detected by the Java Runtime Environment:

SIGSEGV (0xb) at pc=0x00007f5204acf802, pid=3245, tid=0x00007f5041dca700

JRE version: OpenJDK Runtime Environment (8.0_131-b12) (build 1.8.0_131-b12)

Java VM: OpenJDK 64-Bit Server VM (25.131-b12 mixed mode linux-amd64 compressed oops)

Problematic frame:

J 22484 C1 org.velvia.filo.ZeroCopyBinary$class.cachedHash64(Lorg/velvia/filo/ZeroCopyBinary;)J (140 bytes) @ 0x00007f5204acf802 [0x00007f5204acf760+0xa2]

Failed to write core dump. Core dumps have been disabled. To enable core dumping, try "ulimit -c unlimited" before starting Java again

An error report file with more information is saved as:

/root/spark-2.0.0-bin-hadoop2.6/hs_err_pid3245.log

If you would like to submit a bug report, please visit:

http://bugreport.java.com/bugreport/crash.jsp

bin/spark-shell: line 44: 3245 Aborted (core dumped) "${SPARK_HOME}"/bin/spark-submit --class org.apache.spark.repl.Main --name "Spark shell" "$@"

scala> dfResult1.write.format("filodb.spark").option("dataset", "flow_raw").option("row_keys", "exportMs").option("partition_keys", "protocolId").mode(SaveMode.Append).save()
18/05/10 23:09:35 ERROR OneForOneStrategy:
java.lang.NullPointerException
at org.velvia.filo.ZeroCopyBinary$class.cachedHash64(ZeroCopyBinary.scala:77)
at org.velvia.filo.ZeroCopyUTF8String.cachedHash64(ZeroCopyBinary.scala:101)
at org.velvia.filo.ZeroCopyBinary$class.hashCode(ZeroCopyBinary.scala:85)
at org.velvia.filo.ZeroCopyUTF8String.hashCode(ZeroCopyBinary.scala:101)
at java.util.HashMap.hash(HashMap.java:338)
at java.util.HashMap.putIfAbsent(HashMap.java:1061)
at org.velvia.filo.vectors.DictUTF8Vector$.shouldMakeDict(DictUTF8Vector.scala:49)
at org.velvia.filo.vectors.UTF8PtrAppendable.optimizedVector(UTF8Vector.scala:304)
at org.velvia.filo.vectors.UTF8PtrAppendable.suboptimize(UTF8Vector.scala:294)
at org.velvia.filo.vectors.ObjectVector.optimize(ObjectVector.scala:59)

Steps to reproduce
scala> val files=Seq("file:/root/parquetfiles/fragment1523366588968000000.dat","file:/root/parquetfiles/fragment1523350809009000000.dat")
scala> val sqlContext = new org.apache.spark.sql.SQLContext(sc)
scala> val parDF=sqlContext.read.parquet(files:_*)

scala> parDF1.count
res2: Long = 1168530

val dfResult = parDF.withColumn("exporterIp", inetToString(parDF("exporterIp"))).withColumn("srcIp", inetToString(parDF("srcIp"))).withColumn("dstIp", inetToString(parDF("dstIp"))).withColumn("nextHopIp", inetToString(parDF("nextHopIp"))).withColumn("bgpNextHopIp", inetToString(parDF("bgpNextHopIp"))).withColumn("appId", inetToString(parDF("appId"))).withColumn("policyQosClassificationHierarchy", inetToString(parDF("policyQosClassificationHierarchy"))).withColumn("protocolId", parDF("protocolId").cast(IntegerType)).withColumn("srcTos", parDF("srcTos").cast(IntegerType)).withColumn("dstTos", parDF("dstTos").cast(IntegerType)).withColumn("srcMask", parDF("srcMask").cast(StringType)).withColumn("dstMask", parDF("dstMask").cast(StringType)).withColumn("direction", parDF("direction").cast(StringType))

dfResult.write.format("filodb.spark").option("dataset", "flow_metric_raw_new").option("row_keys", "exportMs").option("partition_keys", "protocolId").mode(SaveMode.Append).save()

It throws either JVM errors or null pointer exceptions

Logs

some log

or as attached file (see below)

Unused parts of this template should be removed (including this line).

[PallaviPersistent]

Why does nullpointer exception point to these scala codes
at org.velvia.filo.vectors.DictUTF8Vector$.shouldMakeDict(DictUTF8Vector.scala:46)
at org.velvia.filo.vectors.UTF8PtrAppendable.optimizedVector(UTF8Vector.scala:304)
at org.velvia.filo.vectors.UTF8PtrAppendable.suboptimize(UTF8Vector.scala:294)
at org.velvia.filo.vectors.ObjectVector.optimize(ObjectVector.scala:59)
at org.velvia.filo.BinaryVectorBuilder.toFiloBuffer(BinaryVector.scala:314)
at org.velvia.filo.RowToVectorBuilder$$anonfun$2.apply(RowToVectorBuilder.scala:86)
at org.velvia.filo.RowToVectorBuilder$$anonfun$2.apply(RowToVectorBuilder.scala:86)
at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)
at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)

[velvia]

Hmm, not too sure, but one possible source is multiple partitions trying to write to the same place in Spark. Has the input DataFrame been sorted? Sorry this is on a branch that is not supported and being worked on right now. Current architecture is very different.

…

On May 16 2018, at 10:10 am, PallaviPersistent ***@***.***> wrote: Why does nullpointer exception point to these scala codes at org.velvia.filo.vectors.DictUTF8Vector$.shouldMakeDict(DictUTF8Vector.scala:46) at org.velvia.filo.vectors.UTF8PtrAppendable.optimizedVector(UTF8Vector.scala:304) at org.velvia.filo.vectors.UTF8PtrAppendable.suboptimize(UTF8Vector.scala:294) at org.velvia.filo.vectors.ObjectVector.optimize(ObjectVector.scala:59) at org.velvia.filo.BinaryVectorBuilder.toFiloBuffer(BinaryVector.scala:314) at org.velvia.filo.RowToVectorBuilder$$anonfun$2.apply(RowToVectorBuilder.scala:86) at org.velvia.filo.RowToVectorBuilder$$anonfun$2.apply(RowToVectorBuilder.scala:86) at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234) at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234) — You are receiving this because you are subscribed to this thread. Reply to this email directly, view it on GitHub (#154 (comment)), or mute the thread (https://github.com/notifications/unsubscribe-auth/ABA322ovDEHyEt0FmJ7tQM4vAfgvXkrnks5tzF1ugaJpZM4T6TYi).

[PallaviPersistent]

Thanks for your reply ..
we could write after sorting the data in DataFrame
I have picked up the latest branch from git .. I believe the latest version is 0.7

[PallaviPersistent]

NPE and JVM crash errors are coming intermittently. Sorting the dataframe did help but the error recurred. Also the pid logs generated during JVM crash point to Java thread filo-spark-akka.actor.default-dispatcher-23
Current thread (0x00007fbb0000a800): JavaThread "filo-spark-akka.actor.default-dispatcher-23" [_thread_in_Java, id=18737, stack(0x00007fba51573000,0x00007fba51674000)]

[PallaviPersistent]

hs_err_pid20721.log
Attaching the file generated during JVM crash.