LabeledPoint类的Kryo注册

Question

我试图用Kryo注册运行一个非常简单的scala类。这个类只将数据从文件加载到RDD[LabeledPoint]中。

代码(来自于https://spark.apache.org/docs/latest/mllib-decision-tree.html中的代码)：

import org.apache.spark.{SparkContext, SparkConf}

import org.apache.spark.mllib.linalg.Vectors
import org.apache.spark.mllib.regression.LabeledPoint



object test {
  def main(args: Array[String]) {

    val conf = new SparkConf().setMaster("local").setAppName("test")
    conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
    conf.set("spark.kryo.registrationRequired", "true")
    val sc = new SparkContext(conf)

    sc.getConf.registerKryoClasses(classOf[ org.apache.spark.mllib.regression.LabeledPoint ])
    sc.getConf.registerKryoClasses(classOf[ org.apache.spark.rdd.RDD[org.apache.spark.mllib.regression.LabeledPoint] ])

    // Load data
    val rawData = sc.textFile("data/mllib/sample_tree_data.csv")
    val data = rawData.map { line =>
      val parts = line.split(',').map(_.toDouble)
      LabeledPoint(parts(0), Vectors.dense(parts.tail))
    }

    sc.stop()
    System.exit(0)
  }
}

据我了解，由于我已经设置了spark.kryo.registrationRequired = true，所有已使用的类都必须注册，这样我就注册了RDD[LabeledPoint]和LabeledPoint。

问题

我收到以下错误：

java.lang.IllegalArgumentException: Class is not registered: org.apache.spark.mllib.regression.LabeledPoint[]
Note: To register this class use: kryo.register(org.apache.spark.mllib.regression.LabeledPoint[].class);
    at com.esotericsoftware.kryo.Kryo.getRegistration(Kryo.java:442)
    at com.esotericsoftware.kryo.util.DefaultClassResolver.writeClass(DefaultClassResolver.java:79)
    at com.esotericsoftware.kryo.Kryo.writeClass(Kryo.java:472)
    at com.esotericsoftware.kryo.Kryo.writeClassAndObject(Kryo.java:565)
    at org.apache.spark.serializer.KryoSerializerInstance.serialize(KryoSerializer.scala:162)
    at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:213)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
    at java.lang.Thread.run(Thread.java:745)

据我所知，这意味着类LabeledPoint[]没有注册，而我已经注册了类LabeledPoint。

此外，错误注册类(kryo.register(org.apache.spark.mllib.regression.LabeledPoint[].class);)中提出的代码不起作用。

• 这两门课有什么区别？
• 我怎样才能注册这个班？

Answer 1

非常感谢@eliasah，他指出建议的解决方案(kryo.register(org.apache.spark.mllib.regression.LabeledPoint[].class);)使用的是Java，而不是Scala。

因此，LabeledPoint[]在Scala中的含义是Array[LabeledPoint]。

我通过注册Array[LabeledPoint]类来解决我的问题，即在代码中添加：

sc.getConf.registerKryoClasses(classOf[ Array[org.apache.spark.mllib.regression.LabeledPoint] ])