I'm facing an issue running spark jon on hadoop/yarn cluster it run's fine in local mode but fails in cluster mode due to this null pointer exception
i'm using spark 1.6.2 and scala 2.10.6 both in local and cluster , the application is a streaming application streaming data from kakfa, here is the code where i'm getting the null pointer, i'm able to get data for some of the batches but for some i get the null pointer's as the null pointer's pile up the job fails here is the code snippet of where it fails
I'm new to spark if someone could please point me in the right direction it will be really and helpful
Note :Form the research i have done within map it's iterating over elements of dstream and encountering a null if it next element is null but as per spark documentation empty batches should not cause an issue .
DevMain.scala
Line 1 val lines: DStream[String,Array[Byte]] = myConsumer.createDefaultStream()
Line 2 val keyDeLines = lines.map(lme.aParser);
---------------------------------------------------------------------------
this is createDefaultStream()
def createDefaultStream(): DStream[(String,Array[Byte])] = {
val consumerConfProps = List("zookeeper.connect","group.id","zookeeper.connection.timeout.ms")
val kafkaConf = Utils.getSubProps(props,consumerConfProps)
val topicArray = props.getProperty("topics").split(",")
val topicMap = {
topicArray.map((_, props.getProperty("numthreads").toInt)).toMap
}
KafkaUtils.createStream[String, Array[Byte], StringDecoder, DefaultDecoder](ssc,
kafkaConf,
topicMap,
StorageLevel.MEMORY_ONLY_SER
)
--------------------------------------------------------------------------
this is lme.parser
def aParser(x: (String,Array[Byte])): Option[Map[String,Any]] = {
logInfo("Entered lme: ")
val decodeTry = Injection.invert(x._2)
decodeTry match {
case Failure(e) => {
logInfo(s"Could not decode binary data: " + e.getStackTrace)
None
}
case Success(eventPojo) => {
val bs: String = eventPojo.toString
logInfo("json: " + bs)
}
}
code never enter's the 'lme.aParser' function in the null pointer cases i have put logging on line1 of lme.parser
here is the stacktrace
java.lang.NullPointerException
at DevMain$anonfun$5.apply(DevMain.scala:2)
at scala.collection.Iterator$anon$11.next(Iterator.scala:328)
at scala.collection.Iterator$anon$11.next(Iterator.scala:328)
at scala.collection.Iterator$anon$11.next(Iterator.scala:328)
at scala.collection.Iterator$anon$14.hasNext(Iterator.scala:389)
at org.apache.spark.util.Utils$.getIteratorSize(Utils.scala:1631)
at org.apache.spark.rdd.RDD$anonfun$count$1.apply(RDD.scala:1157)
at org.apache.spark.rdd.RDD$anonfun$count$1.apply(RDD.scala:1157)
at org.apache.spark.SparkContext$anonfun$runJob$5.apply(SparkContext.scala:1858)
at org.apache.spark.SparkContext$anonfun$runJob$5.apply(SparkContext.scala:1858)
at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:66)
at org.apache.spark.scheduler.Task.run(Task.scala:89)
at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:227)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
at java.lang.Thread.run(Thread.java:745)
