将 Spark Dataframe 转换为可变 Map

Question

我是 spark 和 scala 的新手。我正在尝试查询 hive 中的表（从表中选择 2 列）并将生成的数据框转换为 Map。我正在使用 Spark 1.6 和 Scala 2.10.6。

例如：

Dataframe:
+--------+-------+
| address| exists|
+--------+-------+
|address1|   1   |
|address2|   0   |
|address3|   1   |
+--------+-------+ 
should be converted to: Map("address1" -> 1, "address2" -> 0, "address3" -> 1)

这是我正在使用的代码：

val testMap: scala.collection.mutable.Map[String,Any] = Map()
val df= hiveContext.sql("select address,exists from testTable")
qualys.foreach( r => {
  val key = r(0).toString
  val value = r(1)
  testMap+=(key -> value)
  }
)
testMap.foreach(println)

当我运行上面的代码时，我得到了这个错误：

java.lang.NoSuchMethodError: scala.Predef$.ArrowAssoc(Ljava/lang/Object;)Ljava/lang/Object;

它在我尝试将键值对添加到 Map 的行中引发此错误。即testMap+=(key -> value)

我知道使用org.apache.spark.sql.functions.map 有更好、更简单的方法。但是，我使用的是 Spark 1.6，我认为此功能不可用。我尝试了import，但在可用函数列表中没有找到它。

为什么我的方法会出错？ spark 1.6 有没有更好/更优雅的方法来实现这一点？

任何帮助将不胜感激。谢谢！

更新：

我将元素添加到地图的方式更改为：testMap.put(key, value)。

我之前使用+= 来添加元素。现在我不再收到java.lang.NoSuchMethodError 了。但是，没有元素被添加到testMap。在 foreach 步骤完成后，我尝试打印地图的大小以及其中的所有元素，我看到有 zero 元素。

为什么没有添加元素？我也愿意接受任何其他更好的方法。谢谢！！

Answer 1

这可以分为 3 个步骤，每个步骤都已经在 SO 上解决了：

1. 将 DataFrame 转换为 RDD[(String, Int)]
2. 在该 RDD 上调用 collectAsMap() 以获取不可变映射
3. 将该映射转换为可变映射（例如，如 here 所述）

注意：我不知道您为什么需要 mutable 映射 - 值得注意的是，使用 mutable 集合很少有多大意义在斯卡拉。只坚持不可变对象更安全，更容易推理。 “忘记”可变集合的存在使得学习函数式 API（如 Spark 的！）变得更加容易。

Answer 2

您只需从数据框中收集数据并在其上进行迭代，它就会起作用

qualys.collect.map( r => {
val key = r(0).toString
val value = r(1)
testMap+=(key -> value)
 }
)