使用自定义函数时，使用Word2VecModel出现NullPointerException

Question

我正在尝试将word2vec模型对象传递给我的Spark UDF。基本上，我有一个带有电影ID的测试数据集，并且我想将ID与模型对象一起传递，以获取每行的推荐电影数组。

def udfGetSynonyms(model: org.apache.spark.ml.feature.Word2VecModel) = 
     udf((col : String)  => {
          model.findSynonymsArray("20", 1)
})

然而，这给了我一个空指针异常。当我在udf外运行model.findSynonymsArray("20", 1)时，我得到了预期的答案。由于某种原因它不理解udf中的函数，但是可以在udf外运行它。

注意：我在这里添加了“20”，只是为了得到一个固定的答案，看看是否有效。如果我用col替换“20”，它也会出现相同的情况。

感谢您的帮助！

StackTrace:

SparkException: Job aborted due to stage failure: Task 0 in stage 23127.0 failed 4 times, most recent failure: Lost task 0.3 in stage 23127.0 (TID 4646648, 10.56.243.178, executor 149): org.apache.spark.SparkException: Failed to execute user defined function($anonfun$udfGetSynonyms1$1: (string) => array<struct<_1:string,_2:double>>)
at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage2.processNext(Unknown Source)
at org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43)
at org.apache.spark.sql.execution.WholeStageCodegenExec$$anonfun$10$$anon$1.hasNext(WholeStageCodegenExec.scala:614)
at org.apache.spark.sql.execution.collect.UnsafeRowBatchUtils$.encodeUnsafeRows(UnsafeRowBatchUtils.scala:49)
at org.apache.spark.sql.execution.collect.Collector$$anonfun$2.apply(Collector.scala:126)
at org.apache.spark.sql.execution.collect.Collector$$anonfun$2.apply(Collector.scala:125)
at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:87)
at org.apache.spark.scheduler.Task.run(Task.scala:111)
at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:350)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:748)

Caused by: java.lang.NullPointerException
at org.apache.spark.ml.feature.Word2VecModel.findSynonymsArray(Word2Vec.scala:273)
at linebb57ebe901e04c40a4fba9fb7416f724554.$read$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$anonfun$udfGetSynonyms1$1.apply(command-232354:7)
at linebb57ebe901e04c40a4fba9fb7416f724554.$read$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$anonfun$udfGetSynonyms1$1.apply(command-232354:4)
... 12 more

- Roshini

---

编辑后，原始帖子变得大部分无关紧要。你能否编辑问题并清理一下？编辑后 - 这段代码将无法工作，因为 findSynonyms 内部使用了分布式操作。你需要找到另一种方法来解决这个问题。 - zero323

@JoeK 实际上你是正确的，它没有。我已经检查过了，我也无法重现 NPE，请问你可以吗？ - zero323

当然，我会把这个东西整理得更好一些 @user6910411 - Roshini

感谢您的编辑。有趣的是，我无法重现NPE问题。能否也发布一下堆栈跟踪？ - zero323

啊，我想问题可能是因为我有一个集群..我无法在单个机器上重现NPE问题。当在多个节点上运行时，似乎wordVectors可能不可用。 - Roshini

Answer 1

SQL和UDF API有些受限，我不确定是否有一种方法可以将自定义类型用作列或输入到UDF中。经过一番搜索，并没有找到太有用的信息。

相反，您可以使用DataSet或RDD API，只需使用常规Scala函数而不是UDF，例如：

val model: Word2VecModel = ...
val inputs: DataSet[String] = ...
inputs.map(movieId => model.findSynonymsArray(movieId, 10))

或者，我猜你可以将模型序列化为字符串并从中反序列化，但那看起来要丑得多。

- Joe K

---

这里不应该有任何区别。 - zero323

Answer 2

我认为这个问题发生是因为wordVectors是一个瞬态变量

class Word2VecModel private[ml] (
    @Since("1.4.0") override val uid: String,
    @transient private val wordVectors: feature.Word2VecModel)
  extends Model[Word2VecModel] with Word2VecBase with MLWritable {

我通过广播w2vModel.getVectors并在每个分区内重新创建Word2VecModel模型来解决了这个问题。

- nicola007b