Avro4s，如何使用自定义键类型序列化地图？

Question

我正在使用Avro4s。串行化

Map[String, T]

但我现在的情况是

sealed trait Base
case object First extends Base
case object Second extends Base

我需要序列化类似的东西

Map[Base, T]

对于实现这一目标的最佳方法，有什么建议吗？谢谢。

Answer 1

问题是根据阿夫罗规范

映射键被假定为字符串。

因此，Avro支持的唯一类型是Map[String,T]。这意味着您需要编写一些自定义代码，将您的Map[Base, T]映射到Map[String,T]并返回。像这样的东西可能对你有用：

import scala.collection.breakOut
import scala.collection.immutable.Map
import scala.collection.JavaConverters._
import com.sksamuel.avro4s._
import org.apache.avro.Schema
import org.apache.avro.Schema.Field

object BaseMapAvroHelpers {
  private val nameMap: Map[Base, String] = Map(First -> "first", Second -> "second")
  private val revNameMap: Map[String, Base] = nameMap.toList.map(kv => (kv._2, kv._1)).toMap

  implicit def toSchema[T: SchemaFor]: ToSchema[Map[Base, T]] = new ToSchema[Map[Base, T]] {
    override val schema: Schema = Schema.createMap(implicitly[SchemaFor[T]].apply())
  }

  implicit def toValue[T: SchemaFor : ToValue]: ToValue[Map[Base, T]] = new ToValue[Map[Base, T]] {
    override def apply(value: Map[Base, T]): java.util.Map[String, T] = value.map(kv => (nameMap(kv._1), kv._2)).asJava
  }

  implicit def fromValue[T: SchemaFor : FromValue]: FromValue[Map[Base, T]] = new FromValue[Map[Base, T]] {
    override def apply(value: Any, field: Field): Map[Base, T] = {
      val fromValueS = implicitly[FromValue[String]]
      val fromValueT = implicitly[FromValue[T]]
      value.asInstanceOf[java.util.Map[Any, Any]].asScala.map(kv => (revNameMap(fromValueS(kv._1)), fromValueT(kv._2)))(breakOut)
    }
  }
}

用法示例：

case class Wrapper[T](value: T)
def test(): Unit = {
  import BaseMapAvroHelpers._
  val map: Map[Base, String] = Map(First -> "abc", Second -> "xyz")
  val wrapper = Wrapper(map)
  val schema = AvroSchema[Wrapper[Map[Base, String]]]
  println(s"Schema: $schema")

  val bufOut = new ByteArrayOutputStream()
  val out = AvroJsonOutputStream[Wrapper[Map[Base, String]]](bufOut)
  out.write(wrapper)
  out.flush()
  println(s"Avro Out: ${bufOut.size}")
  println(bufOut.toString("UTF-8"))

  val in = AvroJsonInputStream[Wrapper[Map[Base, String]]](new ByteArrayInputStream(bufOut.toByteArray))
  val read = in.singleEntity
  println(s"read: $read")
}

输出结果类似于：

模式：{“类型”：“记录”：“名称”：“包装”，“命名空间”：“所以”，“字段”：{“名称”：“值”，“类型”：{“类型”：“映射”，“值”：“字符串”} Avro : 40 {“值”：{“第一”：“abc”，“第二”：“xyz”}读取:成功(Map(第一次-> abc，第二个-> xyz))