Spark 文档指出,您所要做的就是注册您的类并向 conf 添加两个变量:
import com.esotericsoftware.kryo.Kryo
import org.apache.spark.serializer.KryoRegistrator
class MyRegistrator extends KryoRegistrator {
override def registerClasses(kryo: Kryo) {
kryo.register(classOf[MyClass1])
kryo.register(classOf[MyClass2])
}
}
val conf = new SparkConf().setMaster(...).setAppName(...)
conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
conf.set("spark.kryo.registrator", "mypackage.MyRegistrator")
val sc = new SparkContext(conf)
我已经在我的代码中实现了这个,但是在尝试对 (Text, Text) 的键/值序列文件进行排序时,我仍然遇到序列化错误。我的 MyRegistrator 版本如下所示:
class MyRegistrator extends KryoRegistrator {
override def registerClasses(kryo: Kryo) {
kryo.register(classOf[Text])
}
}
我还在 MyRegistrator 中添加了日志记录,但我没有看到任何日志语句。我还故意拼错了 MyRegistrator 的名称,但作业没有出错。这必须比文档所提供的更多。还有什么我需要做的吗?
我正在使用 Apache Spark 1.0.2。
谢谢
最佳答案
我能够弄清楚如何解决这个问题。我将 Apache Spark 的版本升级到 1.1.0,它开始工作了。我根本没有更改任何代码,唯一更改的是我的 POM。为了证明它有效,我在我的代码中注释掉了所有对 Kryo 的引用并重新运行。它因序列化错误而失败。
关于apache-spark - Apache Spark 中的 Kryo,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/26221031/