GraphFrame中使用AggregateMessages聚合时如何保留所有元素？

嘿，我明白你要做的事儿了——你不是要统计求和或者取首尾这类聚合结果，而是想把边表里每个源顶点对应的所有目标顶点都收集起来，更新到顶点表的downstreamEdges字段对吧？这事儿用Spark的collect_list函数就能轻松搞定，我给你捋清楚步骤：

先明确你的数据结构 #

首先看你给出的顶点表v：

scala> v.show()
+---+---------------+
| id|downstreamEdges|
+---+---------------+
|CCC| null|
|BBB| null|
|QQQ| null|
|DDD| null|
|FFF| null|
|EEE| null|
|AAA| null|
|GGG| null|
+---+---------------+

还有边表e：

scala> e.show()
+---+---+---+
| iD|src|dst|
+---+---+---+
| 1|CCC|AAA|
| 2|CCC|BBB|
...
+---+---+---+

实现聚合收集的代码 #

核心思路是按源顶点src分组，把每个分组下的所有目标顶点dst收集成一个列表，然后关联回顶点表更新字段：

// 先导入需要的Spark函数
import org.apache.spark.sql.functions.{col, collect_list, coalesce}

// 第一步：对边表按src分组，收集所有对应的dst
val aggregatedEdges = e.groupBy(col("src"))
  .agg(collect_list(col("dst")).alias("downstreamEdges"))

// 第二步：和顶点表左关联，保留所有顶点，更新downstreamEdges字段
val updatedVertices = v.join(aggregatedEdges, v("id") === aggregatedEdges("src"), "left")
  .select(
    v("id"),
    // 如果没有关联的边，保留原来的null；也可以换成array()设置为空列表
    coalesce(aggregatedEdges("downstreamEdges"), v("downstreamEdges")).alias("downstreamEdges")
  )

// 查看结果
updatedVertices.show()

补充说明 #

• 如果你需要对收集的目标顶点去重，把collect_list换成collect_set就行，它会自动剔除重复的dst值
• 如果你不需要关联回顶点表，直接使用aggregatedEdges就是按源顶点分组后的所有关联目标顶点列表

内容的提问来源于stack exchange，提问作者Shafique Jamal