最近在使用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
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()
}
}
这里,只是采用Graphx做个demo(很简单啦),来测试Graphx在当前数据量级下的相关性能。实际设备团伙挖掘会更复杂,涉及到各种策略制定。