将包含键值对的数据流转换为 DataStream[ObjectNode] json 到映射 Scala

Question

我正在尝试从 kafka 读取 json 数据并在 Scala.I 中处理它我是 flink 和 kafka 流媒体的新手所以请尝试通过给出解决方案来回答 code.I 想要能够将其转换为包含所有键值对的 Map。

map1.get("FC196") 应该给我休眠，其中 map1 是包含键值对的映射

我面临的挑战是将代码中的 st 变量 DataStream[ObjectNode] 转换为键值对映射。我正在使用 JSonDeserializerSchema.If 我使用简单字符串模式我得到 DataStream[String]。我愿意接受其他建议。

来自kafka的输入格式：

{"FC196":"Dormant","FC174":"A262210940","FC195":"","FC176":"40","FC198":"BANKING","FC175":"AHMED","FC197":"2017/04/04","FC178":"1","FC177":"CBS","FC199":"INDIVIDUAL","FC179":"SYSTEM","FC190":"OK","FC192":"osName","FC191":"Completed","FC194":"125","FC193":"7","FC203":"A10SBPUB000000000004439900053575","FC205":"1","FC185":"20","FC184":"Transfer","FC187":"2","FC186":"2121","FC189":"abcdef","FC200":"afs","FC188":"BR08","FC202":"INDIVIDUAL","FC201":"","FC181":"7:00PM","FC180":"2007/04/01","FC183":"11000000","FC182":"INR"}

代码：

import java.util.Properties
import org.apache.flink.api.scala._
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer09
import org.apache.flink.streaming.util.serialization.SimpleStringSchema



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

    // kafka properties
    val properties = new Properties()
    properties.setProperty("bootstrap.servers", "***.**.*.*:9092")
    properties.setProperty("zookeeper.connect", "***.**.*.*:2181")
    properties.setProperty("group.id", "afs")
    properties.setProperty("auto.offset.reset", "latest")

    val env = StreamExecutionEnvironment.getExecutionEnvironment

    val st = env
      .addSource(new FlinkKafkaConsumer09("new", new JSONDeserializationSchema() , properties))

    st.print()

      env.execute()
  }
}

修改后的代码：

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



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

    case class CC(key: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])

    st.print()

      env.execute()
  }
}

出于某种原因，我无法像您描述的那样在平面图中放置 Try

错误：

INFO [main] (TypeExtractor.java:1804) - No fields detected for class org.json4s.JsonAST$JValue. Cannot be used as a PojoType. Will be handled as GenericType
Exception in thread "main" org.apache.flink.api.common.InvalidProgramException: Task not serializable
    at org.apache.flink.api.scala.ClosureCleaner$.ensureSerializable(ClosureCleaner.scala:172)
    at org.apache.flink.api.scala.ClosureCleaner$.clean(ClosureCleaner.scala:164)
    at org.apache.flink.streaming.api.scala.StreamExecutionEnvironment.scalaClean(StreamExecutionEnvironment.scala:666)
    at org.apache.flink.streaming.api.scala.DataStream.clean(DataStream.scala:994)
    at org.apache.flink.streaming.api.scala.DataStream.map(DataStream.scala:519)
    at org.apache.flink.quickstart.WordCount$.main(WordCount.scala:36)
    at org.apache.flink.quickstart.WordCount.main(WordCount.scala)
Caused by: java.io.NotSerializableException: org.json4s.DefaultFormats$$anon
    at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1184)
    at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548)
    at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509)
    at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432)
    at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178)
    at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548)
    at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509)
    at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432)
    at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178)
    at java.io.ObjectOutputStream.writeObject(ObjectOutputStream.java:348)
    at org.apache.flink.util.InstantiationUtil.serializeObject(InstantiationUtil.java:317)
    at org.apache.flink.api.scala.ClosureCleaner$.ensureSerializable(ClosureCleaner.scala:170)
    ... 6 more

Process finished with exit code 1

Answer 1

这里有两个任务需要处理：

将原始 json 有效负载解析为某种形式的 AST
正在将 AST 转换为您可以使用的格式。

如果您使用 SimpleStringSchema，您可以选择一个不错的 Json 解析器并在一个简单的 flatMap 运算符中解组 json 有效载荷。

您的 build.sbt

的一些依赖项

"org.json4s" %% "json4s-core" % "3.5.1",
"org.json4s" %% "json4s-native" % "3.5.1"

Scala 中有十几个 Json 库可供选择，可以在此处找到一个很好的概述 https://manuel.bernhardt.io/2015/11/06/a-quick-tour-of-json-libraries-in-scala/

然后进行一些解析：

scala> import org.json4s.native.JsonMethods._
import org.json4s.native.JsonMethods._

scala> val raw = """{"key":"value"}"""
raw: String = {"key":"value"}

scala> parse(raw)
res0: org.json4s.JValue = JObject(List((key,JString(value))))

在此阶段，AST 可用。这可以转换为 Map 如下：

scala> res0.values
res1: res0.Values = Map(key -> value)

请记住 Json4s 不执行异常处理，因此这可能会引发异常（当您从 Kafka 获取数据时应避免的事情，它最终会杀死您的工作）。

在 flink 中，这看起来像这样：

env
  .addSource(new FlinkKafkaConsumer09("new", new SimpleStringSchema() , properties))
  .flatMap(raw => Try(JsonMethods.parse(raw).toOption) // this will discard failed instances, you should handle better, ie log
  .map(_.values)

但是，我建议将您的数据表示为一个案例 class。这将需要另一个步骤将 AST 映射到一个案例 class.

在上面的例子中。

scala> implicit val formats = org.json4s.DefaultFormats
formats: org.json4s.DefaultFormats.type = org.json4s.DefaultFormats$@341621da

scala> case class CC(key: String)
defined class CC

scala> parse(raw).extract[CC]
res20: CC = CC(value)

或者在 flink 中：

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

更新：

只需将隐式格式移到 main 方法之外：

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

将包含键值对的数据流转换为 DataStream[ObjectNode] json 到映射 Scala

Converting a Datastream containing key,value pairs as DataStream[ObjectNode] json to a map Scala

scala

apache-kafka

kafka-consumer-api

apache-flink

flink-streaming