将json提取到案例类SCALA FLINK时出错

Question

问题在于在进行案例类提取时映射函数中。case类不可序列化。我隐式地定义了格式DefaultFormats。

package org.apache.flink.quickstart
import java.util.Properties

import com.fasterxml.jackson.databind.{JsonNode, ObjectMapper}
import com.fasterxml.jackson.module.scala.DefaultScalaModule
import org.apache.flink.api.scala._
import org.apache.flink.runtime.state.filesystem.FsStateBackend
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer09
import org.apache.flink.streaming.util.serialization.SimpleStringSchema
import org.json4s.DefaultFormats
import org.json4s._
import org.json4s.native.JsonMethods
import scala.util.Try


case class CC(key:String)

object WordCount{
  def main(args: Array[String]) {

    implicit val formats = org.json4s.DefaultFormats

    // kafka properties
    val properties = new Properties()
    properties.setProperty("bootstrap.servers", "***.**.*.***:9093")
    properties.setProperty("zookeeper.connect", "***.**.*.***:2181")
    properties.setProperty("group.id", "afs")
    properties.setProperty("auto.offset.reset", "earliest")
    val env = StreamExecutionEnvironment.getExecutionEnvironment

   val st = env
       .addSource(new FlinkKafkaConsumer09("new", new SimpleStringSchema() , properties))
       .flatMap(raw => JsonMethods.parse(raw).toOption)
//       .map(_.extract[CC])


    val l = st.map(_.extract[CC])

    st.print()
      env.execute()
  }
}

错误是：

INFO[main]（TypeExtractor.java:1804）-未检测到类组织的字段。json4s。JsonAST$JValue。不能用作PojoType。将作为线程“main”组织中的GenericType异常处理。阿帕奇。Flink。应用程序编程接口。常见的InvalidProgrameException:任务无法在组织中序列化。阿帕奇。Flink。应用程序编程接口。斯卡拉。ClosureCleaner美元。EnsureCleaner.scala:172）可在org。阿帕奇。Flink。应用程序编程接口。斯卡拉。ClosureCleaner美元。clean（ClosureCleaner.scala:164）位于org。阿帕奇。Flink。流动。应用程序编程接口。斯卡拉。拖缆执行环境。scalaClean（streamexecutionenEnvironment.scala:666），网址：org。阿帕奇。Flink。流动。应用程序编程接口。斯卡拉。数据流。clean（DataStream.scala:994）位于org。阿帕奇。Flink。流动。应用程序编程接口。斯卡拉。数据流。地图（DataStream.scala:519）位于org。阿帕奇。Flink。快速启动。WordCount$。main（WordCount.scala:38）位于org。阿帕奇。Flink。快速启动。字数。main（WordCount.scala）由以下原因引起：java。伊奥。NotSerializableException:org。json4s。java上的DefaultFormats$$anon$4。伊奥。ObjectOutputStream。java上的writeObject0（ObjectOutputStream.java:1184）。伊奥。ObjectOutputStream。java上的defaultWriteFields（ObjectOutputStream.java:1548）。伊奥。ObjectOutputStream。在java上编写SerialData（ObjectOutputStream.java:1509）。伊奥。ObjectOutputStream。java上的writeOrdinaryObject（ObjectOutputStream.java:1432）。伊奥。ObjectOutputStream。java上的writeObject0（ObjectOutputStream.java:1178）。伊奥。ObjectOutputStream。java上的defaultWriteFields（ObjectOutputStream.java:1548）。伊奥。ObjectOutputStream。在java上编写SerialData（ObjectOutputStream.java:1509）。伊奥。ObjectOutputStream。java上的writeOrdinaryObject（ObjectOutputStream.java:1432）。伊奥。ObjectOutputStream。java上的writeObject0（ObjectOutputStream.java:1178）。伊奥。ObjectOutputStream。writeObject（ObjectOutputStream.java:348）位于org。阿帕奇。Flink。util。实例化util。org上的serializeObject（InstantiationUtil.java:317）。阿帕奇。Flink。应用程序编程接口。斯卡拉。ClosureCleaner美元。确保可重复使用（ClosureCleaner.scala:170）。。。还有6个

Process finished with exit code 1

Answer 1

解决问题的办法

implicit val formats = org.json4s.DefaultFormats

在主要功能之外，比如

object WordCount{

    implicit val formats = org.json4s.DefaultFormats
    def main(args: Array[String])

或者懒洋洋地初始化格式，比如

implicit lazy val formats = org.json4s.DefaultFormats

里面的主要功能就像

def main(args: Array[String]) {
    implicit lazy val formats = org.json4s.DefaultFormats