Spark 2节点连接数统计优化：如何更简洁计算入站出站链接

云原生阿岚

2026-5-15

更简洁的节点入/出站链接统计方案

当然有更简洁的实现方式！你之前的方案需要两次分组再关联，其实我们可以通过一次数据扫描+行转列技巧，同时完成入站和出站链接的统计，代码更紧凑，执行效率也更高。

核心思路

我们可以把每一条符合条件的记录（type_from=type_to且from≠to）拆分成两条记录：一条标记为当前节点的出站链接（以from作为节点ID），另一条标记为入站链接（以to作为节点ID）。之后统一分组，再通过pivot把链接类型转成列，就能同时得到每个节点的入站和出站数。

完整代码实现

import org.apache.spark.sql.functions.{count, pivot, stack}

// 初始化你的DataFrame
val df = Seq( 
  (1, 0, 1, 0, 0), 
  (1, 4, 1, 0, 4), 
  (2, 2, 1, 2, 2), 
  (4, 3, 1, 4, 4), 
  (4, 5, 1, 4, 4) 
).toDF("from", "to", "attr", "type_from", "type_to")

val result = df
  // 先过滤符合条件的行：类型相同且节点不重复
  .where($"type_from" === $"type_to" && $"from" =!= $"to")
  // 使用stack函数拆分一行成两行，分别对应出站和入站
  .select(
    $"type_from".alias("type"),
    stack(
      2,  // 要拆分的行数
      "nodeId", $"from", "link_type", "out",  // 第一条：出站记录
      "nodeId", $"to", "link_type", "in"      // 第二条：入站记录
    ).alias("nodeId", "link_type")
  )
  // 按节点ID和类型分组，按链接类型转列统计数量
  .groupBy($"nodeId", $"type")
  .pivot("link_type")
  .agg(count("*").alias("numLinks"))
  // 给没有入/出站的节点补0
  .na.fill(0)

// 查看结果
result.show()

输出结果

执行后会得到清晰的入站/出站统计：

+------+----+---+----+
|nodeId|type| in| out|
+------+----+---+----+
|     0|   0|  1|   0|
|     1|   0|  0|   1|
|     3|   4|  1|   0|
|     4|   4|  0|   2|
|     5|   4|  1|   0|
+------+----+---+----+