将 Spark JSON 格式的 RDD 值解析为不同的值答案

【问题标题】：Parse Spark JSON format RDD value into different values将 Spark JSON 格式的 RDD 值解析为不同的值
【发布时间】：2021-03-03 22:42:45
【问题描述】：

我正在尝试在 Spark (Scala) 中对具有 N 个值且一个为 JSON 格式的 RDD 执行某种 flatMap。

例如，当我打印 RDD 时，我有类似的东西：

myRDD.collect().foreach(println)

[2020,{'COL_A': 1064.3667, 'col_B': 14534.2}]
[2020,{'COL_A': 1064.3667, 'col_B': 145.2}]
[2020,{'COL_A': 1064.3667, 'col_B': 15576.2}]

我想要这样的东西：

[2020,1064.3667,14534.2]
[2020,1064.3667,145.2]
[2020,1064.3667,15576.2]

我不知道这是否可以用 flatmap 完成...

谢谢！

【问题讨论】：

标签： json scala apache-spark

【解决方案1】：

使用内置的json4s库解析json。

导入所需的库

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

scala> import org.json4s._
import org.json4s._

scala> val rdd = spark
.sparkContext
.parallelize(
    Seq(
        (2020,"""{"COL_A": 1064.3667, "col_B": 14534.2}"""),
        (2020,"""{"COL_A": 1064.3667, "col_B": 145.2}"""),
        (2020,"""{"COL_A": 1064.3667, "col_B": 15576.2}""")
       )
)

scala> rdd.collect.foreach(println)
(2020,{"COL_A": 1064.3667, "col_B": 14534.2})
(2020,{"COL_A": 1064.3667, "col_B": 145.2})
(2020,{"COL_A": 1064.3667, "col_B": 15576.2})

scala> :paste
// Entering paste mode (ctrl-D to finish)

val transformedRdd = rdd.map { c =>
      implicit val formats = DefaultFormats
      val values = parse(c._2).extract[Map[String,Double]].values.toList
      (c._1,values.head,values.last)
}

// Exiting paste mode, now interpreting.

scala> transformedRdd.collect.foreach(println)
(2020,1064.3667,14534.2)
(2020,1064.3667,145.2)
(2020,1064.3667,15576.2)

【讨论】：