使用 Spark 和 Scala 处理 json RDD 中的每个 json 记录

Question

我需要以下情况的帮助：

我将从 Kafka 获取以下 JSON 格式的数据以触发流式传输

{"id" : 1 , "data" : "AFGH00101219"}
{"id" : 2 , "data" : "AFGH00101215"}
{"id" : 2 , "data" : "AFGH00101216"}
{"id" : 3 , "data" : "AFGH00101218"}

val messages= KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc, kafkaParams, topics)

现在我想处理消息中的每条 JSON 记录，然后每条记录依次返回一组记录。请给我一些想法来完成以下任务。

val output = messages.map(row =>
{
//here I will get each json record. My doubt is how to extract id and data 
//filed values from row and store it into variables.
//Here I need to decode the data filed value which is in hexa decimal format 
//to decimal format.
}

提前致谢。如果问题不清楚，请告诉我。

Answer 1

您可以使用Google GSON 或任何 JSON 解析库，我曾使用如下的 Google GSON 来解析我在 spark 流中收到的 JSON 数据。

// loop each RDD 
lines.foreachRDD(rawRDD => {
    val rdd = rawRDD.filter(!_.isEmpty)
      .map(row => {
        val jobj = new Gson().fromJson(row, classOf[JsonObject])
        val id = jobj.getAsJsonObject("id").getAsString
        val data = jobj.getAsJsonObject("data").getAsString
        // Do something with id and data
      })
  })

另一种方法是从接收到的RDD创建一个Dataframe

lines.foreachRDD(rawRDD => {
  val rdd = rawRDD.filter(!_.isEmpty)
  val df = spark.read.json(rdd)
  df.show(false)
  })

这会从 rdd 创建一个数据框，如下所示，不，您可以将 id 和 data 用于任何其他转换/操作。

+------------+---+
|data        |id |
+------------+---+
|AFGH00101219|1  |
|AFGH00101215|2  |
|AFGH00101216|2  |
|AFGH00101218|3  |
+------------+---+

我希望这会有所帮助！