在ApacheSpark中广播外部库对象

Question

我试图在火花笔记本的阿帕奇火花中做NLP。对于这个特定的例子，我正在使用库https://opennlp.apache.org创建一个块来提取名词短语。由于数据量的增加，我需要转向分布式计算。

问题是我无法广播我的chunker对象。通过阅读文档（只在board上投射数组等简单对象），我尝试了以下方法：

import opennlp.tools.tokenize.WhitespaceTokenizer
import opennlp.tools.cmdline.postag.POSModelLoader
import opennlp.tools.postag.POSTaggerME
import opennlp.tools.chunker.ChunkerModel
import opennlp.tools.chunker.ChunkerME
import java.io.FileInputStream
import java.io.File

//Instantiate the ChunkerME class 
val inputStream = new FileInputStream("fr-chunk.bin"); 
val chunkerModel = new ChunkerModel(inputStream);
val chunkerME = new ChunkerME(chunkerModel); 

val broadCastedChunkerME = sc.broadcast(chunkerME)

但这会引发以下错误：

java.io.NotSerializableException: opennlp.tools.chunker.ChunkerME
Serialization stack:
    - object not serializable (class: opennlp.tools.chunker.ChunkerME, value: opennlp.tools.chunker.ChunkerME@35a5c281)
  at org.apache.spark.serializer.SerializationDebugger$.improveException(SerializationDebugger.scala:40)
  at org.apache.spark.serializer.JavaSerializationStream.writeObject(JavaSerializer.scala:46)
  at org.apache.spark.broadcast.TorrentBroadcast$$anonfun$blockifyObject$2.apply(TorrentBroadcast.scala:268)
  at org.apache.spark.broadcast.TorrentBroadcast$$anonfun$blockifyObject$2.apply(TorrentBroadcast.scala:268)
  at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1303)
  at org.apache.spark.broadcast.TorrentBroadcast$.blockifyObject(TorrentBroadcast.scala:269)
  at org.apache.spark.broadcast.TorrentBroadcast.writeBlocks(TorrentBroadcast.scala:126)
  at org.apache.spark.broadcast.TorrentBroadcast.<init>(TorrentBroadcast.scala:88)
  at org.apache.spark.broadcast.TorrentBroadcastFactory.newBroadcast(TorrentBroadcastFactory.scala:34)
  at org.apache.spark.broadcast.BroadcastManager.newBroadcast(BroadcastManager.scala:56)
  at org.apache.spark.SparkContext.broadcast(SparkContext.scala:1411)
  ... 63 elided

如果我将chunker的初始化封装在函数中，然后在map方法中调用函数，则可以这样做：

def getNPChunks(sentence: String): Array[Chunk] = {
  import opennlp.tools.chunker.ChunkerModel
  import opennlp.tools.chunker.ChunkerME
  import java.io.FileInputStream

  val inputStream = new FileInputStream("fr-chunk.bin"); 
  val chunkerModel = new ChunkerModel(inputStream);

  //Instantiate the ChunkerME class 
  val chunkerME = new ChunkerME(chunkerModel); 

  chunkerME.chunkAsSpans(sentence); 
}

// call the chunker 
line.map(getNPChunks)

但这里的问题是，这段代码效率非常低，因为它正在为rdd中的每个条目初始化chunker对象。因为map函数为rdd的每个条目调用getNPChunks函数，并且为每个条目创建一个新的chunker对象。

由于这种低效的设计，我的spark脚本的运行速度比顺序脚本慢20倍。

我做错了什么？

Answer 1

初始化对象内部scala"chunkerME"，然后广播它。默认情况下，Scala对象是序列化的。编译器对scala对象进行序列化。

或者，如果在scala类中初始化，则需要通过扩展Serializable特性显式序列化scala类。

Answer 2

解决这个问题的方法是使用mapPartitions
这样，您可以为每个分区创建一个chunker，而不是为每行创建一个chunker：

def getChunker():
  val inputStream = new FileInputStream("fr-chunk.bin"); 
  val chunkerModel = new ChunkerModel(inputStream);

  //Instantiate the ChunkerME class 
  val chunkerME = new ChunkerME(chunkerModel); 

line.mapPartitions(it =>
   val chunker = getChunker()
   it.map(line => chunker.chunkAsSpans(line))
)

有关map分区的更多详细信息，请参阅此答案：https://stackoverflow.com/a/39203798/245024