如何取消嵌套具有以下类型的 spark rdd ((String, scala.collection.immutable.Map[String,scala.collection.immutable.Map[String,Int]]))

Question

当我将它打印到屏幕上时，它是一个嵌套地图，内容如下

(5, Map ( "ABCD" -> Map("3200" -> 3,
                    "3350.800" -> 4, 
                    "200.300" -> 3)
 (1, Map ( "DEF" -> Map("1200" -> 32,
                        "1320.800" -> 4, 
                        "2100" -> 3)

我需要这样的东西

Case Class( 5, ABCD 3200, 3)
Case Class(5, ABCD 3350.800, 4)
CaseClass(5,ABCD., 200.300, 3)
CaseClass(1, DEF 1200, 32)
CaseClass(1 DEF, 1320.800, 4)

等等等等。 基本上是一个案例类列表

并将其映射到案例类对象，以便我可以将其保存到 cassandra。 我尝试过做 flatMapValues ，但它只将地图嵌套一层。还使用了 flatMap 。这也不起作用，或者我犯了错误

有什么建议吗？

Answer 1

使用 for-comprehension 和一些模式匹配来解构事物相当简单：

 val in = List((5, Map ( "ABCD" -> Map("3200" -> 3,  "3350.800" -> 4, "200.300" -> 3))),
               (1, Map ("DEF" -> Map("1200" -> 32, "1320.800" -> 4, "2100" -> 3))))

case class Thing(a:Int, b:String, c:String, d:Int)

 for  { (index, m) <- in
        (k,v) <-m
        (innerK, innerV) <- v}
        yield Thing(index, k, innerK, innerV) 

//> res0: List[maps.maps2.Thing] = List(Thing(5,ABCD,3200,3), 
//                                      Thing(5,ABCD,3350.800,4),
//                                      Thing(5,ABCD,200.300,3), 
//                                      Thing(1,DEF,1200,32),
//                                      Thing(1,DEF,1320.800,4),
//                                      Thing(1,DEF,2100,3))

所以让我们挑选一部分进行理解

(index, m) <- in

这是一样的

t <- in
(index, m) = t

在第一行t 将被依次设置为in 的每个元素。 t 因此是一个元组 (Int, Map(...)) 模式匹配让我们将元组的“模式”放在右侧，编译器将元组分开，将index 设置为 Int，将m 设置为 Map。

(k,v) <-m

和之前一样，这等价于

u <-m
(k, v) = u

这一次u 获取 Map 的每个元素。这又是键和值的元组。所以k被依次设置为每个键，v被设置为值。

v 是你的内部地图，所以我们对内部地图再次做同样的事情

(innerK, innerV) <- v}

现在我们拥有了创建案例类所需的一切。 yield 只是说每次通过循环收集“产生”的任何东西。

yield Thing(index, k, innerK, innerV) 

在后台，这只是转换为一组地图/平面地图

yield 就是 Thing(index, k, innerK, innerV) 的值

我们为v 的每个元素获取其中一个

v.map{x=>val (innerK, innerV) = t;Thing(index, k, innerK, innerV)}

但是外部地图的每个元素都有一个内部地图

m.flatMap{y=>val (k, v) = y;v.map{x=>val (innerK, innerV) = t;Thing(index, k, innerK, innerV)}}

（flatMap，因为如果我们刚刚执行map，我们会得到一个列表列表，并且我们希望将其展平为项目列表）

同样，我们对 List 中的每个元素都执行其中一项操作

in.flatMap (z => val (index, m) = z; m.flatMap{y=>val (k, v) = y;v.map{x=>val (innerK, innerV) = t;Thing(index, k, innerK, innerV)}}

让我们在_1、_2 style-y 中这样做。

in.flatMap (z=> z._2.flatMap{y=>y._2.map{x=>;Thing(z._1, y._1, x._1, x._2)}}}

产生完全相同的结果。但是作为理解不是更清楚吗？

Answer 2

如果你喜欢收集操作，你可以这样做

    case class Record(v1: Int, v2: String, v3: Double, v4: Int)

    val data = List(
      (5, Map ( "ABC" ->
        Map(
          3200. -> 3,
          3350.800 -> 4,
          200.300 -> 3))
        ),
      (1, Map ( "DEF" ->
        Map(
          1200. -> 32,
          1320.800 -> 4,
          2100. -> 3))
        )
    )

    val rdd = sc.parallelize(data)

    val result = rdd.flatMap(p => {
      p._2.toList
        .flatMap(q => q._2.toList.map(l => (q._1, l)))
        .map((p._1, _))
    }).map(p => Record(p._1, p._2._1, p._2._2._1, p._2._2._2))

    println(result.collect.toList)
    //List(
    //  Record(5,ABC,3200.0,3),
    //  Record(5,ABC,3350.8,4),
    //  Record(5,ABC,200.3,3),
    //  Record(1,DEF,1200.0,32),
    //  Record(1,DEF,1320.8,4),
    //  Record(1,DEF,2100.0,3)
    //)