Search code examples
scalaapache-sparkstanford-nlp

Stanford LexicalizedParser throws NPE when using in spark

I am attempting to use stanford's LexicalizedParser within Spark RDD map function.

The algorithm is roughly like this:

val parser = LexicalizedParser.loadModel(englishPCFG.ser.gz)
val parserBroadcast = sparkContext.broadcast(parser) // using Kryo serializer here

someSparkRdd.map { case sentence: List[HasWord] =>
    parserBroadcast.value.parse(sentence) //NPE is being thrown see below
}

The reason I would like to instantiate parser once (outside the map) and then just broadcast it, is that the map iterates over almost a million sentences, java garbage collector produces too much overhead and whole processing slows down reasonably.

Upon executing map statement, following NullPointerException is being thrown:

java.lang.NullPointerException
    at edu.stanford.nlp.parser.lexparser.BaseLexicon.isKnown(BaseLexicon.java:152)
    at edu.stanford.nlp.parser.lexparser.BaseLexicon.ruleIteratorByWord(BaseLexicon.java:208)
    at edu.stanford.nlp.parser.lexparser.ExhaustivePCFGParser.initializeChart(ExhaustivePCFGParser.java:1343)
    at edu.stanford.nlp.parser.lexparser.ExhaustivePCFGParser.parse(ExhaustivePCFGParser.java:457)
    at edu.stanford.nlp.parser.lexparser.LexicalizedParserQuery.parseInternal(LexicalizedParserQuery.java:258)
    at edu.stanford.nlp.parser.lexparser.LexicalizedParserQuery.parse(LexicalizedParserQuery.java:536)
    at edu.stanford.nlp.parser.lexparser.LexicalizedParser.parse(LexicalizedParser.java:301)
    at my.class.NounPhraseExtractionWithStanford$$anonfun$extractNounPhrases$3.apply(NounPhraseExtractionWithStanford.scala:28)
    at my.class.NounPhraseExtractionWithStanford$$anonfun$extractNounPhrases$3.apply(NounPhraseExtractionWithStanford.scala:27)
    at scala.collection.TraversableLike$$anonfun$flatMap$1.apply(TraversableLike.scala:251)
    at scala.collection.TraversableLike$$anonfun$flatMap$1.apply(TraversableLike.scala:251)
    at scala.collection.Iterator$class.foreach(Iterator.scala:727)
    at scala.collection.AbstractIterator.foreach(Iterator.scala:1157)
    at scala.collection.IterableLike$class.foreach(IterableLike.scala:72)
    at scala.collection.AbstractIterable.foreach(Iterable.scala:54)
    at scala.collection.TraversableLike$class.flatMap(TraversableLike.scala:251)
    at scala.collection.AbstractTraversable.flatMap(Traversable.scala:105)
    at my.class.NounPhraseExtractionWithStanford$.extractNounPhrases(NounPhraseExtractionWithStanford.scala:27)
    at my.class.HBaseDocumentProducerWithStanford$$anonfun$produceDocumentTokens$3.apply(HBaseDocumentProducerWithStanford.scala:104)
    at my.class.HBaseDocumentProducerWithStanford$$anonfun$produceDocumentTokens$3.apply(HBaseDocumentProducerWithStanford.scala:104)
    at org.apache.spark.rdd.PairRDDFunctions$$anonfun$mapValues$1$$anonfun$apply$15.apply(PairRDDFunctions.scala:674)
    at org.apache.spark.rdd.PairRDDFunctions$$anonfun$mapValues$1$$anonfun$apply$15.apply(PairRDDFunctions.scala:674)
    at scala.collection.Iterator$$anon$11.next(Iterator.scala:328)
    at org.apache.spark.storage.MemoryStore.unrollSafely(MemoryStore.scala:249)
    at org.apache.spark.CacheManager.putInBlockManager(CacheManager.scala:172)
    at org.apache.spark.CacheManager.getOrCompute(CacheManager.scala:79)
    at org.apache.spark.rdd.RDD.iterator(RDD.scala:242)
    at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:35)
    at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:277)
    at org.apache.spark.rdd.RDD.iterator(RDD.scala:244)
    at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:68)
    at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:41)
    at org.apache.spark.scheduler.Task.run(Task.scala:64)
    at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:203)
    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)

In the source code I see that apparently because of many transient class variables of edu.stanford.nlp.parser.lexparser.BaseLexicon the SerDe performed during broadcast (using Kryo serializer) leaves BaseLexicon half-initialized.

I realize that developers of LexParser didn't had spark in mind when designing it but still I would greatly appreciate any hints on how I could use it in my scenario (with spark that is).

Solution

  • One possible workaround, not 100% sure that it'll work:

    class ParseSentence extends (List[HasWord] => WhateverParseReturns) with Serializable {
  def apply(sentence: List[HasWord]) = ParseSentence.parser.parse(sentence)
}

object ParseSentence {
  val parser = LexicalizedParser.loadModel(englishPCFG.ser.gz)
}

someSparkRdd.map(new ParseSentence)

    This way parser shouldn't need to be serialized/deserialized because it isn't captured as a field of the function object.