Как зарегистрировать byte[][] с помощью крио-сериализации для искры

Я пытаюсь полностью использовать крио-сериализацию для искры. Параметр

.set("spark.kryo.registrationRequired", "true")

Это позволит мне узнать, какие классы необходимо зарегистрировать. Я зарегистрировал около 40 классов, некоторые из моих классов и некоторые из классов искры. Я следил за сообщением Require kryo serialization in Spark (Scala), чтобы зарегистрировать/установить все вверх.

Теперь я сталкиваюсь со следующим и не могу понять, как зарегистрировать его в scala. Кто-нибудь решил эту проблему?

Я пробовал кучу разных комбинаций, включая:

kryo.register(classOf[Array[Array[Byte]]])
conf.set("classesToRegister", "classOf[Array[Array[Byte]]]")
conf.registerKryoClasses(Array(classOf[Array[Array[Byte]]]))

Я нашел сообщение без ответа https://mail-archives.apache.org/mod_mbox/spark-user/201603.mbox/%3CCAHCfvsSyUpx78ZFS_A9ycxvtO1=Jp7DfCCAeJKHyHZ1sugqHEQ@mail.gmail.com%3E с той же проблемой.

java.lang.RuntimeException: com.esotericsoftware.kryo.KryoException: java.lang.IllegalArgumentException: Class is not registered: byte[][]
Note: To register this class use: kryo.register(byte[][].class);
Serialization trace:
buffers (org.apache.spark.sql.columnar.CachedBatch)
at com.esotericsoftware.kryo.serializers.FieldSerializer$ObjectField.write(FieldSerializer.java:585)
at com.esotericsoftware.kryo.serializers.FieldSerializer.write(FieldSerializer.java:213)
at com.esotericsoftware.kryo.Kryo.writeClassAndObject(Kryo.java:568)
at org.apache.spark.serializer.KryoSerializationStream.writeObject(KryoSerializer.scala:158)
at org.apache.spark.serializer.SerializationStream.writeAll(Serializer.scala:153)
at org.apache.spark.storage.BlockManager.dataSerializeStream(BlockManager.scala:1190)
at org.apache.spark.storage.BlockManager.dataSerialize(BlockManager.scala:1199)
at org.apache.spark.storage.MemoryStore.getBytes(MemoryStore.scala:191)
at org.apache.spark.storage.BlockManager.doGetLocal(BlockManager.scala:480)
at org.apache.spark.storage.BlockManager.getBlockData(BlockManager.scala:302)
at org.apache.spark.network.netty.NettyBlockRpcServer$$anonfun$2.apply(NettyBlockRpcServer.scala:57)
at org.apache.spark.network.netty.NettyBlockRpcServer$$anonfun$2.apply(NettyBlockRpcServer.scala:57)
at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:244)
at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:244)
at scala.collection.IndexedSeqOptimized$class.foreach(IndexedSeqOptimized.scala:33)
at scala.collection.mutable.ArrayOps$ofRef.foreach(ArrayOps.scala:108)
at scala.collection.TraversableLike$class.map(TraversableLike.scala:244)
at scala.collection.mutable.ArrayOps$ofRef.map(ArrayOps.scala:108)
at org.apache.spark.network.netty.NettyBlockRpcServer.receive(NettyBlockRpcServer.scala:57)
at org.apache.spark.network.server.TransportRequestHandler.processRpcRequest(TransportRequestHandler.java:114)
at org.apache.spark.network.server.TransportRequestHandler.handle(TransportRequestHandler.java:87)
at org.apache.spark.network.server.TransportChannelHandler.channelRead0(TransportChannelHandler.java:101)
at org.apache.spark.network.server.TransportChannelHandler.channelRead0(TransportChannelHandler.java:51)
at io.netty.channel.SimpleChannelInboundHandler.channelRead(SimpleChannelInboundHandler.java:105)
at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:308)
at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:294)
at io.netty.handler.timeout.IdleStateHandler.channelRead(IdleStateHandler.java:266)
at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:308)
at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:294)
at io.netty.handler.codec.MessageToMessageDecoder.channelRead(MessageToMessageDecoder.java:103)
at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:308)
at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:294)
at io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:244)
at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:308)
at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:294)
at io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:846)
at io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:131)
at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:511)
at io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:468)
at io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:382)
at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:354)
at io.netty.util.concurrent.SingleThreadEventExecutor$2.run(SingleThreadEventExecutor.java:111)
at java.lang.Thread.run(Thread.java:745)

person John Engelhart    schedule 13.06.2016    source источник


Ответы (2)


conf.registerKryoClasses(Array( Class.forName("[[B"))) 

должно сработать

person Harel Gliksman    schedule 13.06.2016
comment
Это полезно. У меня была такая же проблема, и это помогло мне ее решить. Однако я не понимаю, как [[B соответствует байту[][]. Я знаю, что это старый пост, но если бы кто-то мог дать более глубокое объяснение, это было бы полезно. - person Tim Ryan; 29.09.2016
comment
@TimRyan См. docs.oracle .com/javase/6/docs/api/java/lang/ - person Reinstate Monica; 03.11.2016

Я знаю этот чрезвычайно старый вопрос, но для тех, кто может искать решение для любого массива классов:

case class Person(name: String)

conf.registerKryoClasses(Array(
   classOf[Array[Array[Person]]]
))

Или в случае этого конкретного вопроса:

conf.registerKryoClasses(Array(
   classOf[Array[Array[Byte]]]
))
person Noam Shaish    schedule 15.04.2019
comment
это не сработало @Noam Shaish, у меня val personList: Array[Person] = (1 to 100000).map(value=> Person(value+"",value)).toArray и .registerKryoClasses(Array( classOf[Array[Array[Person]]] )) не сработало, пишет `Вызвано: java.lang.IllegalArgumentException: класс не зарегистрирован: KyroExample$Person Примечание: для регистрации этого класса используйте: kryo. зарегистрироваться(KyroExample$Person.class);` . У меня есть полный вопрос, если вы можете посмотреть stackoverflow.com/questions/59099104/ - person supernatural; 29.11.2019