如何在akka-kryo-serializer中使用Scala映射

Question

package entellect.spike.Kryo

import java.io.{ByteArrayInputStream, ByteArrayOutputStream}

import com.esotericsoftware.kryo.Kryo
import com.esotericsoftware.kryo.io.{Input, Output}


object KryoSpike extends App {

  val kryo = new Kryo()
  kryo.addDefaultSerializer(classOf[scala.collection.Map[_,_]], classOf[ScalaMapSerializer])
  kryo.addDefaultSerializer(classOf[scala.collection.generic.MapFactory[scala.collection.Map]], classOf[ScalaMapSerializer])

  val testin = Map("id" -> "objID", "field1" -> "field1Value")

  val outStream = new ByteArrayOutputStream()
  val output = new Output(outStream, 4096)
  kryo.writeClassAndObject(output, testin)
  output.flush()


  val input = new Input(new ByteArrayInputStream(outStream.toByteArray), 4096)
  val testout = kryo.readObject(input, classOf[Map[String,String]])

  println(testout.toString)

}

下面的行不编译，因为编译没有找到“hashmap$hashtriemap”

classOf[ScalaImmutableMapSerializer])
    kryo.register(classOf[scala.collection.immutable.HashMap$HashTrieMap],

40）

最后，我的示例如下所示：

object KryoSpike extends App {

  val kryo = new Kryo()
  kryo.addDefaultSerializer(classOf[scala.collection.Map[_,_]], classOf[ScalaImmutableMapSerializer])
  kryo.addDefaultSerializer(classOf[scala.collection.generic.MapFactory[scala.collection.Map]], classOf[ScalaImmutableMapSerializer])

  val testin = Map("id" -> "objID", "field1" -> "field1Value")

  val outStream = new ByteArrayOutputStream()
  val output = new Output(outStream, 4096)
  kryo.writeClassAndObject(output, testin)
  output.flush()


  val input = new Input(new ByteArrayInputStream(outStream.toByteArray), 4096)
  val testout = kryo.readObject(input, classOf[Map[String,String]])

  println(testout.toString)

}

  "org.apache.spark" % "spark-core_2.11" % "2.3.1",
  "org.apache.spark" % "spark-sql_2.11" % "2.3.1",
  "com.typesafe.akka" %% "akka-stream" % "2.5.16",
  "com.typesafe.akka" %% "akka-http-spray-json" % "10.1.4",
  "com.typesafe.akka" %% "akka-stream-kafka" % "0.22",
  "com.github.romix.akka" %% "akka-kryo-serialization" % "0.5.0"

注我没有使用Kryo特定的Akka特性，我使用它作为一个通用的序列化框架。火花也一样。没有直接插入火花或akka配置。

Answer 1

解决方案使用

带映射的ScalaImmutableAbstractMapSerializer

以及writeObject和readObject方法。

package entellect.spike.Kryo

import java.io.{ByteArrayInputStream, ByteArrayOutputStream}

import com.esotericsoftware.kryo.Kryo
import com.esotericsoftware.kryo.io.{Input, Output}
import com.romix.scala.serialization.kryo._

object KryoSpike extends App {


  val kryo = new Kryo()
  kryo.setRegistrationRequired(false)
  kryo.addDefaultSerializer(classOf[scala.collection.Map[_,_]], classOf[ScalaImmutableAbstractMapSerializer])
  kryo.addDefaultSerializer(classOf[scala.collection.generic.MapFactory[scala.collection.Map]], classOf[ScalaImmutableAbstractMapSerializer])

  val testin = Map("id" -> "objID", "field1" -> "field1Value")

  val outStream = new ByteArrayOutputStream()
  val output = new Output(outStream, 4096)
  kryo.writeObject(output, testin)
  output.flush()


  val input = new Input(new ByteArrayInputStream(outStream.toByteArray), 4096)
  val testout = kryo.readObject(input, classOf[scala.collection.Map[_,_]])

  println(testout.toString)

}