使用 GraphX 从二分网络创建单分图

Question

我正在试验 Spark，而且我是 Scala 和 GraphX 的新手。

我正在建立一个由电影和在其中扮演角色的演员组成的双向网络。 我的输入是一个空格分隔的文件，每行一个边，第一列是电影 ID，第二列是演员 ID，这是一个摘录：

movie actor
1 1 
2 1 
2 2 
3 1 
3 3 
3 4 
3 5 
3 6 
3 7 
4 1 

我的目标是研究演员之间的关系。为此，我想创建一个单部分图 act​​or-actor，其边缘值取决于这些演员的常见电影数量。

我从创建边的 RDD 开始：

val edges: RDD[Edge[String]] = sc.textFile("file:///home/actorMovie").map { line =>
    val fields = line.split(" ")
    Edge(fields(0).toLong, fields(1).toLong)
  }

看起来不错：

edges.take(10)
res8: Array[org.apache.spark.graphx.Edge[String]] = Array(Edge(1,1,null), Edge(2,1,null), Edge(2,2,null), Edge(3,1,null), Edge(3,3,null), Edge(3,4,null), Edge(3,5,null), Edge(3,6,null), Edge(3,7,null), Edge(4,1,null))

然后从那里开始，我不知道如何将其转换为加权单部图。

Answer 1

您现在拥有的不是二分图。要使其成为有效的图表，您必须进行大量转换。最简单的方法是使用数据框：

val df = spark.read
  .option("delimiter", " ") // if there is a header
  .option("header", "true")
  .csv(path)
  .toDF("movie", "actor")

一旦你有这样的数据自加入和聚合：

import org.apache.spark.sql.functions._

val e = df.alias("df1")
  .join(df.alias("df2"), "movie")
  // Sort soruce and destination
  .groupBy(
     greatest("df1.actor", "df2.actor").as("srcId"), 
     least("df1.actor", "df2.actor").as("dstId"))
  .agg(count("*").as("attr"))

这可以转换为Graph 或GraphFrame，正如我在这里展示的How to obtain convert DataFrame to specific RDD?。

要创建有效的图表，您必须对标签进行编码。例如：

val dfe = df.select(
  concat(lit("m"), $"movie").as("movie"),
  concat(lit("a"), $"actor").as("actor"))

然后您可以按照How to encode string values into numeric values in Spark DataFrame 中的说明应用StringIndexer 或使用GraphFrame 自动执行此操作：

import graphframes._

val gf = GraphFrame.fromEdges(dfe.toDF("src", "dst"))

然后您可以使用消息传递或图形模式匹配来查找两个跃点邻居。