在 spark-shell 和 Scala 罐子中使用 Kryo 注册复杂的 scala 类

Question

我有一个新的 spark 2.3.1 应用程序...它运行有一段时间没问题，但现在随着数据量的增加它坏了。

原始错误是 kryo 序列化问题...com.esotericsoftware.kryo.KryoException: java.lang.NegativeArraySizeException 失败。最奇怪的部分是它不一致......如果我运行相同的代码，相同的数据，在我的非共享集群上，它可能会或可能不会失败并且看起来完全运行dom.

我已将 spark.kryoserializer.buffer.max 从 256m（我的默认值）增加到 2047m（最大值）只是为了看看会发生什么，它仍然失败并出现相同的错误。我还尝试增加失败的 RDD 中的并行度（每个执行程序 6x 从 3x），但没有运气。

现在我正在尝试运行 spark-shell --conf spark.kryo.registrationRequired=true 中的代码片段以找到我需要注册以在序列化时缩小大小的所有 classes，然后逐渐添加它们到 --conf 'spark.kryo.classesToRegister=org.myOrg.MyClass1,org.myOrg.MyClass2' 并稍后将它们移入 jar (conf.registerKryoClasses(Array(classOf[MyClass1], classOf[MyClass2]))) 之后我找到它们（比我预期的要多）。

有一个我完全不知道如何注册。错误看起来像这样...

Caused by: java.lang.IllegalArgumentException: Class is not registered: org.myOrg.MyClass[]
Note: To register this class use: kryo.register(org.myOrg.MyClass[].class);

我怀疑这是 Iterable[MyClass] 对其他 class 的争论 class MyOuterClass(val mcs: Iterable[MyClass]) 但我尝试注册的所有内容都无法正常工作。我相信 MyClass[] 是 java.lang.Array[MyClass] 但我已经尝试注册我能想到的 Array、Iterable、[] 等的每个组合，并且注册失败。

对于在 spark-shell 的命令行启动和最终在代码中注册 Iterable、List、TupleN 的语法有什么建议吗？最终我也会有一些非常嵌套的元组，但我还没有做到这一点。

我在 Whosebug 中可以找到的最接近的结果在这里，但我也无法使它对我有用。

提前致谢。

编辑

澄清一下...在成功注册后 MyClass 我仍然收到一条错误消息 Class is not registered: MyClass[] 我无法弄清楚 [] 最后是什么或如何注册让那些消失。

Answer 1

如果您 class 的名字是 MyClass 然后尝试注册 [LMyClass;

conf.registerKryoClasses(Array( Class.forName("[LMyClass;")))

它应该为 MyClass

加载和注册数组 class

在 spark-shell 和 Scala 罐子中使用 Kryo 注册复杂的 scala 类

Registering complex scala classes with Kryo in spark-shell and Scala jars

java

serialization

scala

kryo

apache-spark