scala - 在spark中组合/连接ID行

Question

给定一个充满 ID 行的文件，例如

i1, i2, i5
i3, i4
i2, i6, i7
i4, i8
i9, i3

您如何通过链接相同的 ID 来加入他们？因此，对于上面的示例，第 1 行通过 i2 链接到第 3 行，第 2 行分别通过 i4 和 i3 链接到第 4 行和第 5 行。这将为您提供以下内容(删除重复项)

i1, i2, i5, i6, i7
i3, i4, i8, i9

我可以通过循环遍历行来做到这一点，但想知道您将如何以功能方式进行处理？

Answer 1

当您使用 Apache Spark 时，您可以使用内置的 GraphX 组件来为您完成这项工作。

import org.apache.spark.graphx._

def cross[Y](xs: Traversable[Y], ys: Traversable[Y]) = for { x <- xs; y <- ys } yield (x, y)

val data = sc.parallelize(List(
    "1\t5\t3", 
    "3\t9\t30",
    "7\t10\t12",
    "10\t7\t13"
))

val prep = data.map(x => x.split("\t").map(_.toLong).toList)

val vertex = prep
  .flatMap(x => x)
  .map(x => x -> s"ID=$x")

val edges = prep
  .map(x => cross(x, x))
  .flatMap(x => x)
  .map(x => new Edge(x._1, x._2, "likes"))

val graph = Graph(vertex, edges)
val linked = graph
  .connectedComponents
  .vertices
  .map(_.swap)
  .groupByKey

linked.take(10).foreach(println)

将打印出以下结果:

(1,CompactBuffer(30, 3, 9, 1, 5))
(7,CompactBuffer(7, 10, 12, 13))

Cross 只是创建两个列表的叉积，以便我们可以在所有顶点之间创建边。

connectedComponents 函数将遍历图并找到共享边的所有顶点并创建一个新图，其中每个顶点都是 Vertex Id ->“Primary”Vertex ID 的元组。

所以:

graph.connectedComponents.vertices.take(10).foreach(println)

打印出来

(30,1)
(1,1)
(3,1)
(5,1)
(7,7)
(9,1)
(10,7)
(12,7)
(13,7)

如您所见，1 和 7 已被选为“主顶点”并链接到第一个图中的所有连接顶点。因此，简单的交换和分组会将所有连接的 id 组合在一起。