###spark版### Spark Graphx 进行团伙的识别（community detection）

本文主要是介绍###spark版### Spark Graphx 进行团伙的识别（community detection），希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

最近在使用Spark Graphx，拿Graphx做了点实验。对大规模图常见的分析方法有连通图挖掘，团伙挖掘等。在金融科技领域，尤其风控领域，会有各种重要的关联网络，并且这种网络图十分庞大。所以，Spark Graphx这种分布式计算框架十分适合这种场景。下面以设备间关联网络（节点数亿级别）为例，采用Graphx做一个设备团伙挖掘demo。团伙识别的算法采用的是Graphx自带的LabelPropagation算法。

下面的是Graphx示例代码(仅仅是demo)：

其中输入文件格式：

A B weight

备注（A，B 代表设备id，String类型，weight：int，关联代表权重）

因为Graphx节点类型只支持Long，不支持String，所以，需要进行相应的转换，这里用到的广播变量进行idmap。

github链接： https://github.com/dylan-fan/spark_graphx_community_detection

[java] view plain copy

package com.org.test
import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
import org.apache.spark.rdd.RDD
import org.apache.spark.graphx._
import scala.collection.mutable.Set
object DeviceCom {
def main(args: Array[String]) {
if (args.length < 3) {
println("usage: spark-submit com.org.test.DeviceCom <input> <output> <iternum>")
System.exit(1)
}
val conf = new SparkConf()
conf.setAppName("DeviceCom-" + System.getenv("USER"))
val sc = new SparkContext(conf)
val input = args(0)
val output = args(1)
val iternum = args(2).toInt
val vids = sc.textFile(input)
.flatMap(line => line.split("\t").take(2))
.distinct
.zipWithUniqueId()
.map(x => (x._1, x._2.toLong))
val vids_map = sc.broadcast(vids.collectAsMap())
val vids_rdd = vids.map {
case (username, userid) =>
(userid, username)
}
val raw_edge = sc.textFile(input)
.map(line => line.split("\t"))
val col = raw_edge.collect()
val edges_rdd = sc.parallelize(col.map {
case (x) =>
(vids_map.value(x(0)), vids_map.value(x(1)))
})
val g = Graph.fromEdgeTuples(edges_rdd, 1)
val lp = lib.LabelPropagation.run(g, iternum).vertices
val LpByUsername = vids_rdd.join(lp).map {
case (id, (username, label)) =>
(username, label)
}
LpByUsername.map(x => x._1 + "\t" + x._2).saveAsTextFile(output)
sc.stop()
}
}