在 Spark 中将 Dataframe 转换为 Map(Key-Value)

Question

所以，我在 Spark 中有一个如下所示的 DataFrame：

它有 30 列：只显示其中的一部分！

[ABCD,color,NORMAL,N,2015-02-20,1]
[XYZA,color,NORMAL,N,2015-05-04,1]
[GFFD,color,NORMAL,N,2015-07-03,1]
[NAAS,color,NORMAL,N,2015-08-26,1]
[LOWW,color,NORMAL,N,2015-09-26,1]
[KARA,color,NORMAL,N,2015-11-08,1]
[ALEQ,color,NORMAL,N,2015-12-04,1]
[VDDE,size,NORMAL,N,2015-12-23,1]
[QWER,color,NORMAL,N,2016-01-18,1]
[KDSS,color,NORMAL,Y,2015-08-29,1]
[KSDS,color,NORMAL,Y,2015-08-29,1]
[ADSS,color,NORMAL,Y,2015-08-29,1]
[BDSS,runn,NORMAL,Y,2015-08-29,1]
[EDSS,color,NORMAL,Y,2015-08-29,1]

所以，我必须将此数据帧转换为 Scala 中的键值对，使用键作为数据帧中的某些列，并将唯一值分配给从索引 0 到计数的键（不同的键数） .

例如：使用上面的例子，我想在 Scala 中的 map(key-value) 集合中有一个输出，如下所示：

    ([ABC_color_NORMAL_N_1->0]
    [XYZA_color_NORMAL_N_1->1]
    [GFFD_color_NORMAL_N_1->2]
    [NAAS_color_NORMAL_N_1->3]
    [LOWW_color_NORMAL_N_1->4]
    [KARA_color_NORMAL_N_1->5]
    [ALEQ_color_NORMAL_N_1->6]
    [VDDE_size_NORMAL_N_1->7]
    [QWER_color_NORMAL_N_1->8]
    [KDSS_color_NORMAL_Y_1->9]
    [KSDS_color_NORMAL_Y_1->10]
    [ADSS_color_NORMAL_Y_1->11]
    [BDSS_runn_NORMAL_Y_1->12]
    [EDSS_color_NORMAL_Y_1->13]
    )

我是 Scala 和 Spark 的新手，我尝试过这样做。

 var map: Map[String, Int] = Map()
    var i = 0
    dataframe.foreach( record =>{
    //Is there a better way of creating a key!
        val key = record(0) + record(1) + record(2) + record(3)
        var index = i
        map += (key -> index)
        i+=1
          }
        )

但是，这不起作用。：/完成后地图为空。

Answer 1

您的代码中的主要问题是试图修改在 workers 上执行的代码中在驱动程序端创建的变量。使用 Spark 时，您可以将 RDD 转换中的驱动程序端变量仅用作“只读”值。

具体来说：

• 地图是在驱动机器上创建的
• 地图（及其初始空值）被序列化并发送到工作节点
• 每个节点都可能更改地图（本地）
• 当foreach 完成时，结果会被丢弃 - 结果不发送回驱动程序。

要解决这个问题 - 您应该选择一个返回更改后的 RDD（例如 map）的转换来创建密钥，使用 zipWithIndex 添加正在运行的“ids”，然后使用 collectAsMap 获取所有数据作为地图返回给驱动程序：

val result: Map[String, Long] = dataframe
  .map(record => record(0) + record(1) + record(2) + record(3))
  .zipWithIndex()
  .collectAsMap()

至于密钥创建本身 - 假设您要包含前 5 列，并在它们之间添加分隔符 (_)，您可以使用：

record => record.toList.take(5).mkString("_")