用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 

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

你现在得到的不是二部图。要使它成为一个有效的图，您必须进行大量的转换。做到这一点最简单的方法是使用Dataframes：

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，正如我在这里所展示的，如何获得将DataFrame转换为特定的rdd？。

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

import graphframes._

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

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