GraphXを試してみた - teruuuのブログ

GraphXとは

Apache Sparkのライブラリの一つでグラフの並列分散処理を行うことができます。Sparkの基本APIや他のライブラリの機能と合わせて利用することができるため、入力データの事前処理からグラフの生成、分析まで一貫して行えるのが強みとなっています。用途としてはtwitterなどの影響力や、レコメンドなどに利用できるのかと思います。

実際に動かしてみる

では実際に動かして試してみたいと思います。ここのチュートリアルを試してみたいと思います。

事前準備

build.sbtのlibraryDependenciesに以下を追加しておいてください。

“org.apache.spark” % “spark-graphx_2.11” % sparkVersion,

GraphXについて

GraphXではノードとエッジで構成される以下のGraphクラスを使用して分析を行います。

class Graph[VD, ED] {
  val vertices: VertexRDD[VD]
  val edges: EdgeRDD[ED]
}

例えば、ノードにユーザ、エッジにユーザ間の関連を表す場合は以下のようにGraphを表すことができます。

val users: RDD[(VertexId, (String, String))] =
  sc.parallelize(Array((3L, ("rxin", "student")), (7L, ("jgonzal", "postdoc")),
                       (5L, ("franklin", "prof")), (2L, ("istoica", "prof"))))

 // Create an RDD for edges
 val relationships: RDD[Edge[String]] =
   sc.parallelize(Array(Edge(3L, 7L, "collab"),    Edge(5L, 3L, "advisor"),
                        Edge(2L, 5L, "colleague"), Edge(5L, 7L, "pi")))
 // Define a default user in case there are relationship with missing user
 val defaultUser = ("John Doe", "Missing")
 // Build the initial Graph
 val graph = Graph(users, relationships, defaultUser)

Graphに対して3つめの引数を渡しておりますが、これは関連情報はあるけど実際の人物が見つからなかった時に使用するデフォルトユーザになります。

グラフのトリプレット情報を出力する

先ほどのユーザと関連で構成されるグラフの情報を出力する場合はtripletsを使用します。tripletsはノードとエッジを組み合わせた情報になりまして今回のサンプルであればfranklinはrxinのadvisorと言った情報を表しています。

// トリプレットでグラフの情報を表示
println("show graph")
val facts: RDD[String] =
  graph.triplets.map(triplet =>
    triplet.srcAttr._1 + " is the " + triplet.attr + " of " + triplet.dstAttr._1)
facts.collect.foreach(println(_))

上記処理で以下のトリプレット情報が出力されます。

rxin is the collab of jgonzal
franklin is the advisor of rxin
istoica is the colleague of franklin
franklin is the pi of jgonzal

簡単なフィルター処理を行ってみる

グラフのトリプレットについてはフィルター処理を簡単に行うことができまして、例えばadvisorの関係のみを出力したい場合は以下のようになります。

val facts2: RDD[String] =
  graph.triplets.filter(t => t.attr equals  "advisor").map(triplet =>
    triplet.srcAttr._1 + " is the " + triplet.attr + " of " + triplet.dstAttr._1)
facts2.collect.foreach(println(_))

それからユーザ指定で誰かに対する関係の情報のみを出したい場合は以下のようになります。

println("show filterd by node graph")
val facts3: RDD[String] =
  graph.triplets.filter(t => t.srcAttr._1 equals  "franklin").map(triplet =>
    triplet.srcAttr._1 + " is the " + triplet.attr + " of " + triplet.dstAttr._1)
facts3.collect.foreach(println(_))

中心性分析

GraphXではどれだけ中心に位置するか、SNSであればどれだけ影響力があるかを調べるための中心性分析も行うことができます。ここではGraphXのリポジトリにあるサンプルデータを利用して影響力のある人物を調べてみたいと思います。使用するのは以下のuser情報とフォロー情報になります。
◯users.text

1,BarackObama,Barack Obama
2,ladygaga,Goddess of Love
3,jeresig,John Resig
4,justinbieber,Justin Bieber
6,matei_zaharia,Matei Zaharia
7,odersky,Martin Odersky
8,anonsys

◯followers.text

エッジ情報のファイルを読み込んでから中心性を分析は以下の処理だけで行えます。手軽そうです。

// Load the edges as a graph
val graphRank = GraphLoader.edgeListFile(sc, "data/followers.txt")
// Run PageRank
val ranks = graphRank.pageRank(0.0001).vertices

それから、ユーザ情報を読み込んで中心性解析結果を付与するのは以下になります。

val usersRank = sc.textFile("data/users.txt").map { line =>
  val fields = line.split(",")
  (fields(0).toLong, fields(1))
}
val ranksByUsername = usersRank.join(ranks).map {
  case (id, (username, rank)) => (username, rank)
}

あとは中心性情報を付与したユーザ情報を一覧出力します。

println(ranksByUsername.collect().mkString("\n"))

そしたらこんな感じの結果が得られると思います。

(justinbieber,0.15)
(matei_zaharia,0.7013599933629602)
(ladygaga,1.390049198216498)
(BarackObama,1.4588814096664682)
(jeresig,0.9993442038507723)
(odersky,1.2973176314422592)

これをみると誰からもフォローされていなjustinbieberが低くなっていて、2名からフォローされているBarackObama,odersky,jeresigの影響力が高くなっていることが確認できます。気になったのが2名にフォローされているjeresigよりもladygagaの方が数値が高い点です。graphRank.pageRankに渡した引数も影響している気もするのですが、2名にフォローされているBarackObamaに唯一フォローされているということで影響力が高いと判定されているのでしょうか。
試しにfollowers.txtを以下のように修正してみると

以下のような結果が得られladygagaの影響力が下がってjustinbieberの影響力が上がっているのか確認できました。想定通りです。

(justinbieber,0.7697996414951923)
(matei_zaharia,0.7013599933629602)
(ladygaga,0.7697996414951923)
(BarackObama,1.4583520976357462)
(jeresig,0.9993442038507723)
(odersky,1.2973176314422592)

使ってみた感想

簡単な分析を行う分には手軽で良さそうな気がしました。