如何有效地将 DataFrame 转换为（列名-> 列值列表）的 Map答案

【问题标题】：How to turn a DataFrame into a Map of (column name -> list of column values) efficiently如何有效地将 DataFrame 转换为（列名-> 列值列表）的 Map
【发布时间】：2019-05-17 17:01:12
【问题描述】：

我想要实现的是，对于以下 DataFrame：

-------------------------
| FOO   | BAR   | BAZ   |
| lorem | ipsum | dolor |
| sit   | amet  | dolor |
| lorem | lorem | dolor |
-------------------------

生成以下输出：

Map(
 FOO -> List("lorem", "sit"),
 BAR -> List("ipsum", "amet", "lorem"),
 BAZ -> List("dolor")
)

这是我想出的 Scala 代码：

val df = data.distinct

df.columns.map((key) => {
  val distinctValues = df
    .select(col(key))
    .collect
    .map(df => df.getString(0))
    .toList
    .distinct
  (key, distinctValues)
}).toMap

我已经尝试过使用 RDD 来替代此代码，不知何故，它们的速度提高了大约 30%，但问题仍然存在：这一切都非常低效。

我在本地 Cassandra 上运行 Spark，该 Cassandra 托管只有 1000 行的示例数据集，但这些操作会生成大量日志，并且需要 7 秒以上才能完成。

我是不是做错了什么，有更好的方法吗？

【问题讨论】：

df.select(df.columns map (c => collect_set(c) as c): _*).first.getValuesMap[Seq[String]](df.columns) 将是一个小的改进，但总体思路是不可扩展的，并且在通用集群上的 Spark 中通常会出现第二个延迟。
我认为这可能是一个重复的问题 - stackoverflow.com/questions/37949494/…
@user6910411 确实提高了很多性能，谢谢！您能否详细说明为什么这是不可扩展的？因为输出会变得太大？
好吧。您收集所有唯一值，并转换为本地结构，以便当且仅当最终结果足够小以由每个节点（驱动程序和执行程序）在内存中处理时它才能工作。如果您先验地知道基数很小，那么可以工作，但是假设有人在那里放了一个随机实数:) 如果您愿意放宽您的要求，您可以melt 框架，然后只取不同的值来保持分布。跨度>

标签： scala apache-spark dataframe apache-spark-sql rdd

【解决方案1】：

您有一个名为 collect_set 的函数

df.select(collect_set($"FOO"), collect_set($"BAR"), collect_set($"BAZ"))

【讨论】：

【解决方案2】：

识别列的唯一值

for (x<- df.columns)
{
df.select(x).groupBy(x).count
}

我认为使用 approx distinct 会更快。

import org.apache.spark.sql.functions.approx_count_distinct
df.agg(approx_count_distinct("some_column"))

这是 zero323 的一篇很好的帖子，解释了这一点>>

How to count occurrences of each distinct value for every column in a dataframe?

【讨论】：

我想你可能误解了我的问题，因为我不是要计算不同的值，而是要将所有这些不同的值放入一个序列中。
你想加快这个过程 - 一种方法是通过用大约 distinct 替换它来提高 distinct 本身的速度，这就是我所暗示的
@Leothorn 但这会很重要，而 Bertrand 想要独特的价值。