flink如何将trait反序列化为case

ig9co6j1 于 2021-06-21 发布在 Flink

关注(0)|答案(1)|浏览(404)

我用的是flink的（1.7.2）kafka消费者。如何反序列化几个扩展相同特性的case类？例如

import spray.json.{DefaultJsonProtocol, RootJsonFormat}
  trait Foo
  case class Boo(name: String) extends Foo
  case class Buzz(name: String, age: Int) extends Foo
  object Formats extends DefaultJsonProtocol{
  implicit val booFormat: RootJsonFormat[Boo] = 
       jsonFormat1(Boo.apply)
  implicit val buzzFormat: RootJsonFormat[Buzz] = 
       jsonFormat2(Buzz.apply)
}

我用的是Kafka消费者 DeserializationSchema :

class FooSchema extends DeserializationSchema[Foo]{
  @transient lazy val log = LoggerFactory.getLogger(this.getClass)

  implicit val typeInfo = createTypeInformation[Foo]

  override def deserialize(bytes: Array[Byte]): Foo = {
      val foo = new String(bytes, StandardCharsets.UTF_8).parseJson
        .convertTo[Foo] //doesn't compile, I need to deserialize to Boo and Buzz
      log.debug(s"Received Boo")
      foo

  }
  override def isEndOfStream(t: Foo): Boolean = false
  override def getProducedType: TypeInformation[Foo] = createTypeInformation[Foo]
}

任何想法都将不胜感激

scala apache-flink flink-streaming spray-json

来源：https://stackoverflow.com/questions/55783151/flink-how-to-deserialize-trait-to-case

1条答案

按热度按时间

kmb7vmvb1#

请尝试使用shapeless，它可以自动导出ADT的解码器，如下所示：

sealed trait Foo
case class Boo(name: String) extends Foo
case class Buzz(name: String, age: Int) extends Foo

object MyFormats extends DefaultJsonProtocol with FamilyFormats {
  implicit val formats = shapeless.cachedImplicit[JsonFormat[Foo]]
}

别忘了做这个 sealed . 注意，原始json需要包含 type 字段消歧器 rawJsonString.parseJson.convertTo[Foo] 比如说工作

object Main extends App {
    import MyFormats._

    val rawJsonBuzz =
      """
        |{
        |  "name": "Picard",
        |  "age": 60,
        |  "type": "Buzz"
        |}
      """.stripMargin

    val buzz = rawJsonBuzz.parseJson.convertTo[Foo]
    println(buzz)

    val rawJsonBoo =
      """
        |{
        |  "name": "Picard",
        |  "type": "Boo"
        |}
      """.stripMargin

    val boo = rawJsonBoo.parseJson.convertTo[Foo]
    println(boo)
  }

应输出

Buzz(Picard,60)
Boo(Picard)

赞(0）回复(0）举报 2021-06-21

我来回答

flink如何将trait反序列化为case

1条答案

相关问题

热门标签

最新问答