在 groupby 之后将 Spark DataFrame 的行聚合到 String

Posted 2023-04-15

【中文标题】在 groupby 之后将 Spark DataFrame 的行聚合到 String

【英文标题】：Aggregate rows of Spark DataFrame to String after groupby

【发布时间】：2017-07-04 16:05:55

【问题描述】：

我是 Spark 和 Scale 的新手，可能真的需要提示来解决我的问题。所以我有两个 DataFrames A（列 id 和名称）和 B（列 id 和文本）想加入它们，按 id 分组并将所有文本行组合成一个字符串：

一个

+--------+--------+
|      id|    name|
+--------+--------+
|       0|       A|
|       1|       B|
+--------+--------+

B

+--------+ -------+
|      id|    text|
+--------+--------+
|       0|     one|
|       0|     two|
|       1|   three|
|       1|    four|
+--------+--------+

想要的结果：

+--------+--------+----------+
|      id|    name|     texts|
+--------+--------+----------+
|       0|       A|   one two|
|       1|       B|three four|
+--------+--------+----------+

到目前为止，我正在尝试以下方法：

var C = A.join(B, "id")
var D = C.groupBy("id", "name").agg(collect_list("text") as "texts")

这很好用，除了我的文本列是字符串数组而不是字符串。非常感谢您的帮助。

【参考方案1】：

我只是在你的中添加一些小功能来提供正确的解决方案，即

A.join(B, Seq("id"), "left").orderBy("id").groupBy("id", "name").agg(concat_ws(" ", collect_list("text")) as "texts")

【参考方案2】：

很简单：

val bCollected = b.groupBy('id).agg(collect_list('text).as("texts")
val ab = a.join(bCollected, a("id") == bCollected("id"), "left")

第一个 DataFrame 是即时结果，b DataFrame 为每个 id 收集了文本。然后你将加入它。 bCollected 应该小于 b 本身，因此它可能获得更好的随机播放时间

【讨论】：

这个问题在某处被骗了。