GraphX Advent Calendar - Day 07 - VertexRDD

f:id:teppei-studio:20100306105411j:plain

昨日は、VertexRDD と EdgeRDD の概要について書きました。今日はVertexRDDが持つ昨日をひとつずつ見て行きたいと思います。

頂点を特定の条件で抽出するための関数です。

val filteredVertices:VertexRDD[(String, Long)]
  = vertices.filter{ case (vid:VertexId, (name:String, value:Long)) => value > 150 }

Day05 でも紹介した、それぞれの頂点の編集処理を行うための関数です。

val mappedVertices:VertexRDD[Long]
  = vertices.mapValues( (vid:VertexId, attr:(String, Long) ) => attr._2 * attr._1.length)

※ diffはどうにも正しく稼働してくれないです。バグなんじゃないかと思い、issue あげています
[SPARK-4600] org.apache.spark.graphx.VertexRDD.diff does not work - ASF JIRA

Vertices に Vertices に JOIN させるための関数です。JOINさせるVerticesの方が足らないことがあり得ることを想定している場合に利用する関数で、ミスマッチする場合に代替的にセットする値を、 getOrElse で指定することができます。

val leftJoinedVertices = vertices.leftJoin(verticesWithCountry){
  (vid, left, right) => (left._1, left._2, right.getOrElse("World"))
}

leftJoin と同じく、Vertices に Vertices に JOIN させるための関数です。JOINさせるVerticesの方が足らないことがあり得ることを想定していない場合に利用する関数です。

val innerJoinedVertices = vertices.innerJoin(verticesWithCountry){
     (vid, left, right) => (left._1, left._2, right)
}

頂点情報について、IDの重複があるデータがあった場合に、それをグラフにJOINさせるには、重複を排除してあげる必要があります。

例えば、以下のような頂点があった場合に、

1,Taro,100
2,Jiro,200
3,Sabo,300

以下のような情報をJOINさせる場合です。

1,10
2,20
2,21
3,30
3,31

これを、例えば、以下のようにID毎に集約してからJOINさせる必要があります。

1,10
2,41
3,61

この場合、 aggregateUsingIndex を使います。

val auiVertices:VertexRDD[Long] = vertices.aggregateUsingIndex(verticesWithNum, _ + _)

明日はEdgeRDDについてです。