在spark-shell和Scala jars中用Kryo注册复杂的scala类。

Posted 2021-03-28

我有一个新的火花2.3.1应用程序......它运行正常了一段时间，但现在它的破坏，因为数据量已经增加。

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

我已经增加了 spark.kryoserializer.buffer.max 以至于 2047m 从 256m (我的默认值)，只是想看看会发生什么，结果还是以同样的错误失败。 我也试过增加失败的 RDD 的并行度（每个执行器从 3x 增加到 6x），但没有成功。

现在，我尝试在以下程序中运行代码片段 spark-shell --conf spark.kryo.registrationRequired=true 找到我需要注册的所有类，以便在序列化时缩小尺寸，然后将它们递增添加到 --conf 'spark.kryo.classesToRegister=org.myOrg.MyClass1,org.myOrg.MyClass2' 并将在以后把它们移到罐子里 (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 MyOuterClass(val mcs: Iterable[MyClass]) 但我尝试注册的一切都失败了。 我相信 MyClass[] 是一个 java.lang.Array[MyClass] 但我已经尝试注册了每一个组合的 Array, Iterable, []等我能想到的，都没能注册成功。

有什么建议可以让我们用什么语法来注册 Iterable, List, TupleN 在命令行启动spark-shell和最终在代码中？ 最终我也会有一些非常嵌套的元组，但我还没有达到那个程度。

我在stackoverflow中找到的最接近的结果是在这里，但我也不能让这个工作对我来说。 要求在Spark(Scala)中使用kryo序列化。

先谢谢你了。

编辑

我想说明一下......在成功注册后 MyClass 我仍然得到一个错误信息 Class is not registered: MyClass[] 我不知道什么是 [] 是在最后或如何注册，使这些去。

答案

如果你的类名是 MyClass 然后尝试注册 [LMyClass;

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

它应该加载并注册数组类，用于 MyClass