Spark ML Kmeans提供:org.apache.spark.SparkException:无法执行用户定义的函数($ anonfun $ 2 :(向量)=> int)

问题描述

我尝试加载KmeansModel，然后从中取出标签:

I try to load the KmeansModel and then get the label out of it :

这是我编写的代码:

 val kMeansModel = KMeansModel.load(trainedMlModel.mlModelFilePath)
              val arrayOfElements = measurePoint.measurements.map(a => a._2).toSeq
              println(s"ArrayOfELements::::$arrayOfElements")
              val arrayDF = sparkContext.parallelize(arrayOfElements).toDF()
              arrayDF.show()
              val vectorDF = new VectorAssembler().setInputCols(arrayDF.columns).setOutputCol("features").transform(arrayDF)
              vectorDF.printSchema()
              vectorDF.show()
              val loadedModel = kMeansModel.setFeaturesCol("features").setPredictionCol("label")
              val labelDF = loadedModel.transform(vectorDF)
              labelDF.printSchema()
              labelDF.show()
              val label = labelDF.rdd.map(_.getAs[Int]("label")).collect().head

它生成的错误StackTrace在这里:

The Error StackTrace that it generates is here :

org.apache.spark.SparkException: Failed to execute user defined function($anonfun$2: (vector) => int)
        at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIterator.processNext(Unknown Source)
        at org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43)
        at org.apache.spark.sql.execution.WholeStageCodegenExec$$anonfun$8$$anon$1.hasNext(WholeStageCodegenExec.scala:377)
        at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:408)
        at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:408)
        at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:408)
        at scala.collection.Iterator$class.foreach(Iterator.scala:893)
        at scala.collection.AbstractIterator.foreach(Iterator.scala:1336)
        at scala.collection.generic.Growable$class.$plus$plus$eq(Growable.scala:59)
        at scala.collection.mutable.ArrayBuffer.$plus$plus$eq(ArrayBuffer.scala:104)
        at scala.collection.mutable.ArrayBuffer.$plus$plus$eq(ArrayBuffer.scala:48)
        at scala.collection.TraversableOnce$class.to(TraversableOnce.scala:310)
        at scala.collection.AbstractIterator.to(Iterator.scala:1336)
        at scala.collection.TraversableOnce$class.toBuffer(TraversableOnce.scala:302)
        at scala.collection.AbstractIterator.toBuffer(Iterator.scala:1336)
        at scala.collection.TraversableOnce$class.toArray(TraversableOnce.scala:289)
        at scala.collection.AbstractIterator.toArray(Iterator.scala:1336)
        at org.apache.spark.rdd.RDD$$anonfun$collect$1$$anonfun$13.apply(RDD.scala:935)
        at org.apache.spark.rdd.RDD$$anonfun$collect$1$$anonfun$13.apply(RDD.scala:935)
        at org.apache.spark.SparkContext$$anonfun$runJob$5.apply(SparkContext.scala:1944)
        at org.apache.spark.SparkContext$$anonfun$runJob$5.apply(SparkContext.scala:1944)
        at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:87)
        at org.apache.spark.scheduler.Task.run(Task.scala:99)
        at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:282)
        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)
    Caused by: java.lang.IllegalArgumentException: requirement failed
        at scala.Predef$.require(Predef.scala:212)
        at org.apache.spark.mllib.util.MLUtils$.fastSquaredDistance(MLUtils.scala:486)
        at org.apache.spark.mllib.clustering.KMeans$.fastSquaredDistance(KMeans.scala:589)
        at org.apache.spark.mllib.clustering.KMeans$$anonfun$findClosest$1.apply(KMeans.scala:563)
        at org.apache.spark.mllib.clustering.KMeans$$anonfun$findClosest$1.apply(KMeans.scala:557)
        at scala.collection.mutable.ArraySeq.foreach(ArraySeq.scala:74)
        at org.apache.spark.mllib.clustering.KMeans$.findClosest(KMeans.scala:557)
        at org.apache.spark.mllib.clustering.KMeansModel.predict(KMeansModel.scala:59)
        at org.apache.spark.ml.clustering.KMeansModel.predict(KMeans.scala:134)
        at org.apache.spark.ml.clustering.KMeansModel$$anonfun$2.apply(KMeans.scala:125)
        at org.apache.spark.ml.clustering.KMeansModel$$anonfun$2.apply(KMeans.scala:125)
        ... 27 more
    17/03/08 23:11:41 WARN TaskSetManager: Lost task 3.0 in stage 26.0 (TID 45, localhost, executor driver): org.apache.spark.SparkException: Failed to execute user defined function($anonfun$2: (vector) => int)
        at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIterator.processNext(Unknown Source)
        at org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43)
        at org.apache.spark.sql.execution.WholeStageCodegenExec$$anonfun$8$$anon$1.hasNext(WholeStageCodegenExec.scala:377)
        at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:408)
        at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:408)
        at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:408)
        at scala.collection.Iterator$class.foreach(Iterator.scala:893)
        at scala.collection.AbstractIterator.foreach(Iterator.scala:1336)
        at scala.collection.generic.Growable$class.$plus$plus$eq(Growable.scala:59)
        at scala.collection.mutable.ArrayBuffer.$plus$plus$eq(ArrayBuffer.scala:104)
        at scala.collection.mutable.ArrayBuffer.$plus$plus$eq(ArrayBuffer.scala:48)
        at scala.collection.TraversableOnce$class.to(TraversableOnce.scala:310)
        at scala.collection.AbstractIterator.to(Iterator.scala:1336)
        at scala.collection.TraversableOnce$class.toBuffer(TraversableOnce.scala:302)
        at scala.collection.AbstractIterator.toBuffer(Iterator.scala:1336)
        at scala.collection.TraversableOnce$class.toArray(TraversableOnce.scala:289)
        at scala.collection.AbstractIterator.toArray(Iterator.scala:1336)
        at org.apache.spark.rdd.RDD$$anonfun$collect$1$$anonfun$13.apply(RDD.scala:935)
        at org.apache.spark.rdd.RDD$$anonfun$collect$1$$anonfun$13.apply(RDD.scala:935)
        at org.apache.spark.SparkContext$$anonfun$runJob$5.apply(SparkContext.scala:1944)
        at org.apache.spark.SparkContext$$anonfun$runJob$5.apply(SparkContext.scala:1944)
        at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:87)
        at org.apache.spark.scheduler.Task.run(Task.scala:99)
        at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:282)
        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)
    Caused by: java.lang.IllegalArgumentException: requirement failed
        at scala.Predef$.require(Predef.scala:212)
        at org.apache.spark.mllib.util.MLUtils$.fastSquaredDistance(MLUtils.scala:486)
        at org.apache.spark.mllib.clustering.KMeans$.fastSquaredDistance(KMeans.scala:589)
        at org.apache.spark.mllib.clustering.KMeans$$anonfun$findClosest$1.apply(KMeans.scala:563)
        at org.apache.spark.mllib.clustering.KMeans$$anonfun$findClosest$1.apply(KMeans.scala:557)
        at scala.collection.mutable.ArraySeq.foreach(ArraySeq.scala:74)
        at org.apache.spark.mllib.clustering.KMeans$.findClosest(KMeans.scala:557)
        at org.apache.spark.mllib.clustering.KMeansModel.predict(KMeansModel.scala:59)
        at org.apache.spark.ml.clustering.KMeansModel.predict(KMeans.scala:134)
        at org.apache.spark.ml.clustering.KMeansModel$$anonfun$2.apply(KMeans.scala:125)
        at org.apache.spark.ml.clustering.KMeansModel$$anonfun$2.apply(KMeans.scala:125)
        ... 27 more

推荐答案

这意味着新数据和用于训练模型的数据之间存在尺寸不匹配.您的代码失败MLUtils.scala:486 处的nofollow noreferrer>，它检查两个Vectors是否具有相同的大小:

This means there is dimension mismatch between new data and data used to train the model. Your code fails at MLUtils.scala:486 which checks if both Vectors have the same size:

private[mllib] def fastSquaredDistance(
    v1: Vector,
    norm1: Double,
    v2: Vector,
    norm2: Double,
    precision: Double = 1e-6): Double = {
  val n = v1.size
  require(v2.size == n)
  ...
}

这篇关于Spark ML Kmeans提供:org.apache.spark.SparkException:无法执行用户定义的函数($ anonfun $ 2 :(向量)=> int)的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持！