Spark - 将 Map 转换为单行 DataFrame

Question

在我的应用程序中，我需要从一个Map创建一个单行DataFrame。

例如，对于以下Map：

("col1" -> 5, "col2" -> 10, "col3" -> 6)

这将被转换为一个DataFrame，其中只有一行数据，映射键将成为列名称。

col1 | col2 | col3
5    | 10   | 6

如果你想知道为什么我需要这个 - 我只需要使用MongoSpark连接器将一个带有一些统计数据的文档保存到MongoDB中，该连接器允许保存DFs和RDDs。

- Daniil Andreyevich Baunov

---

当您尝试在Spark中并行化它时会发生什么？ - OneCricketeer

键是有序的吗，还是您想按字母顺序排序？ - Andrey Tyukin

@AndreyTyukin 不，顺序无关紧要。 - Daniil Andreyevich Baunov

@cricket_007，我认为并行化对于Maps不起作用。 - Daniil Andreyevich Baunov

Answer 1

我认为对列名进行排序不会有任何影响。

  import org.apache.spark.sql.types._
  val map = Map("col1" -> 5, "col2" -> 6, "col3" -> 10)
  val (keys, values) = map.toList.sortBy(_._1).unzip
  val rows = spark.sparkContext.parallelize(Seq(Row(values: _*)))
  val schema = StructType(keys.map(
    k => StructField(k, IntegerType, nullable = false)))
  val df = spark.createDataFrame(rows, schema)
  df.show()

给出：

+----+----+----+
|col1|col2|col3|
+----+----+----+
|   5|   6|  10|
+----+----+----+

这个想法很简单：将地图转换为元组列表，解压缩，将键转换为模式，值转换为单行RDD，从这两个部分构建数据框架（ createDataFrame 接口的使用方法 有点奇怪，可以接受 java.util.List 和其他参数，但出于某种原因不接受通常的 scala List）。

- Andrey Tyukin

---

我正在使用Scala 2.11，因此在上面的map.toList.sortBy(_._1).unzip中（我认为）无法编译：toList不是map的成员，._1不是数字... 有什么想法如何解决这个问题吗？ - David Urry

Answer 2

这是您需要的：

val map: Map[String, Int] = Map("col1" -> 5, "col2" -> 6, "col3" -> 10)

val df = map.tail
  .foldLeft(Seq(map.head._2).toDF(map.head._1))((acc,curr) => acc.withColumn(curr._1,lit(curr._2)))


df.show()

+----+----+----+
|col1|col2|col3|
+----+----+----+
|   5|   6|  10|
+----+----+----+

- Raphael Roth

Answer 3

一个对Rapheal答案的小变化。你可以创建一个虚拟列DF（1*1），然后使用foldLeft添加映射元素，最后删除虚拟列。这样，你的foldLeft就变得简单明了，容易记忆。

val map: Map[String, Int] = Map("col1" -> 5, "col2" -> 6, "col3" -> 10)

val f = Seq("1").toDF("dummy")

map.keys.toList.sorted.foldLeft(f) { (acc,x) => acc.withColumn(x,lit(map(x)) ) }.drop("dummy").show(false)

+----+----+----+
|col1|col2|col3|
+----+----+----+
|5   |6   |10  |
+----+----+----+

- stack0114106