对于关联性较强的图,找出来的连通图非常大,这时串行化的极大团算法,仍然会耗时很久,这里利用剪枝的思想减少样本数据量,但是对于大图,优化空间有限。

Spark Graphx 实现图中极大团挖掘, 伪并行化算法（spark graphx图算法案例） 挖掘 伪并行化 第1张

####背景:####

Spark Graphx并未提供极大团挖掘算法

当下的极大团算法都是串行化的算法,基于Bron–Kerbosch算法

####思路:####

spark graphx提供了连通图的算法,连通图和极大团都是无向图中的概念,极大团为连通图的子集

利用spark graphx 找出连通图,在从各个连通图中,利用串行化的极大团算法,找出极大团 (伪并行化)

对于关联性较强的图,找出来的连通图非常大,这时串行化的极大团算法,仍然会耗时很久,这里利用剪枝的思想减少样本数据量,但是对于大图,优化空间有限

期待真正的并行化的极大团算法

####配置文件:####

  1. graph_data_path=hdfs://localhost/graph_data
  2. out_path=hdfs://localhost/clique
  3. ck_path=hdfs://localhost/checkpoint
  4. numIter=50剪枝次数
  5. count=3极大团顶点数大小
  6. algorithm=2极大团算法,1:个人实现2:jgrapht
  7. percent=90剪枝后的顶点数,占前一次的百分比,如果剪完后,还剩下90%的数据,那么剪枝效率已然不高
  8. spark.master=local
  9. spark.app.name=graph
  10. spark.serializer=org.apache.spark.serializer.KryoSerializer
  11. spark.yarn.executor.memoryOverhead=20480
  12. spark.yarn.driver.memoryOverhead=20480
  13. spark.driver.extraJavaOptions=-XX:+UseG1GC-XX:+UseCompressedOops-XX:+DisableExplicitGC
  14. spark.executor.extraJavaOptions=-XX:+UseG1GC-XX:+UseCompressedOops-XX:+DisableExplicitGC
  15. spark.driver.maxResultSize=10g
  16. spark.default.parallelism=60

jgrapht

####样本数据:####

{"src":"0","dst":"1"} {"src":"0","dst":"2"} {"src":"0","dst":"3"} {"src":"1","dst":"0"} {"src":"2","dst":"1"} {"src":"3","dst":"5"} {"src":"4","dst":"6"} {"src":"5","dst":"4"} {"src":"6","dst":"5"} {"src":"3","dst":"2"} {"src":"2","dst":"3"} {"src":"6","dst":"4"} {"src":"3","dst":"4"} {"src":"4","dst":"3"} {"src":"2","dst":"6"} {"src":"6","dst":"2"} {"src":"6","dst":"7"} {"src":"7","dst":"6"}

####样本图:####

Spark Graphx 实现图中极大团挖掘, 伪并行化算法（spark graphx图算法案例） 挖掘 伪并行化 第2张

####输出:####

0,1,2 0,2,3 3,4,5 4,5,6

####代码实现:####

  1. importjava.utilimportjava.util.Properties
  1. importorg.apache.spark.broadcast.Broadcast
  2. importorg.apache.spark.graphx.{Edge,Graph}
  3. importorg.apache.spark.rdd.RDD
  4. importorg.apache.spark.sql.{Row,SQLContext}
  5. importorg.apache.spark.storage.StorageLevel
  6. importorg.apache.spark.{SparkConf,SparkContext}
  7. importorg.jgrapht.alg.BronKerboschCliqueFinder
  8. importorg.jgrapht.graph.{DefaultEdge,SimpleGraph}
  10. importscala.collection.JavaConverters._
  11. importscala.collection.mutable
  13. objectApplicationTitan{
  14. defmain(args:Array[String]){
  15. valprop=newProperties()
  16. prop.load(getClass.getResourceAsStream("/config.properties"))
  18. valgraph_data_path=prop.getProperty("graph_data_path")
  19. valout_path=prop.getProperty("out_path")
  20. valck_path=prop.getProperty("ck_path")
  21. valcount=Integer.parseInt(prop.getProperty("count"))
  22. valnumIter=Integer.parseInt(prop.getProperty("numIter"))
  23. valalgorithm=Integer.parseInt(prop.getProperty("algorithm"))
  24. valpercent=Integer.parseInt(prop.getProperty("percent"))
  25. valconf=newSparkConf()
  26. try{
  27. Runtime.getRuntime.exec("hdfsdfs-rm-r"+out_path)
  28. //Runtime.getRuntime.exec("cmd.exe/Crd/s/q"+out_path)
  29. }catch{
  30. caseex:Exception=>
  31. ex.printStackTrace(System.out)
  32. }
  34. prop.stringPropertyNames().asScala.foreach(s=>{
  35. if(s.startsWith("spark")){
  36. conf.set(s,prop.getProperty(s))
  37. }
  38. })
  39. conf.registerKryoClasses(Array(getClass))
  40. valsc=newSparkContext(conf)
  41. sc.setLogLevel("ERROR")
  42. sc.setCheckpointDir(ck_path)
  43. valsqlc=newSQLContext(sc)
  44. try{
  45. vale_df=sqlc.read
  46. //.json(graph_data_path)
  47. .parquet(graph_data_path)
  49. vare_rdd=e_df
  50. .mapPartitions(it=>{
  51. it.map({
  52. caseRow(dst:String,src:String)=>
  53. valsrc_long=src.toLong
  54. valdst_long=dst.toLong
  55. if(src_long<dst_long)(src_long,dst_long)else(dst_long,src_long)
  56. })
  57. }).distinct()
  58. e_rdd.persist(StorageLevel.MEMORY_AND_DISK_SER)
  60. varbc:Broadcast[Set[Long]]=null
  61. variter=0
  62. varbc_size=0
  63. //剪枝
  64. while(iter<=numIter){
  65. valtemp=e_rdd
  66. .flatMap(x=>List((x._1,1),(x._2,1)))
  67. .reduceByKey((x,y)=>x+y)
  68. .filter(x=>x._2>=count-1)
  69. .mapPartitions(it=>it.map(x=>x._1))
  70. valbc_value=temp.collect().toSet
  71. bc=sc.broadcast(bc_value)
  72. e_rdd=e_rdd.filter(x=>bc.value.contains(x._1)&&bc.value.contains(x._2))
  73. e_rdd.persist(StorageLevel.MEMORY_AND_DISK_SER)
  74. iter+=1
  75. if(bc_size!=0&&bc_value.size>=bc_size*percent/100){
  76. println("totaliter:"+iter)
  77. iter=Int.MaxValue
  78. }
  79. bc_size=bc_value.size
  80. }
  82. //构造图
  83. valedge:RDD[Edge[Long]]=e_rdd.mapPartitions(it=>it.map(x=>Edge(x._1,x._2)))
  84. valgraph=Graph.fromEdges(edge,0,StorageLevel.MEMORY_AND_DISK_SER,StorageLevel.MEMORY_AND_DISK_SER)
  86. //连通图
  87. valcc=graph.connectedComponents().vertices
  88. cc.persist(StorageLevel.MEMORY_AND_DISK_SER)
  90. cc.join(e_rdd)
  91. .mapPartitions(it=>it.map(x=>((math.random*10).toInt.toString.concat(x._2._1.toString),(x._1,x._2._2))))
  92. .aggregateByKey(List[(Long,Long)]())((list,v)=>list:+v,(list1,list2)=>list1:::list2)
  93. .mapPartitions(it=>it.map(x=>(x._1.substring(1),x._2)))
  94. .aggregateByKey(List[(Long,Long)]())((list1,list2)=>list1:::list2,(list3,list4)=>list3:::list4)
  95. .filter(x=>x._2.size>=count-1)
  96. .flatMap(x=>{
  97. if(algorithm==1)
  98. find(x,count)
  99. else
  100. find2(x,count)
  101. })
  102. .mapPartitions(it=>{
  103. it.map({
  104. caseset=>
  105. vartemp=""
  106. set.asScala.foreach(x=>temp+=x+",")
  107. temp.substring(0,temp.length-1)
  108. case_=>
  109. })
  110. })
  111. //.coalesce(1)
  112. .saveAsTextFile(out_path)
  113. }
  115. catch{
  116. caseex:Exception=>
  117. ex.printStackTrace(System.out)
  118. }
  119. sc.stop()
  120. }
  121. //自己实现的极大团算法
  122. deffind(x:(String,List[(Long,Long)]),count:Int):mutable.Set[util.Set[String]]={
  123. println(x._1+"|s|"+x._2.size)
  124. println("BKCliqueFinder---"+x._1+"---"+System.currentTimeMillis())
  125. valneighbors=newutil.HashMap[String,util.Set[String]]
  126. valfinder=newCliqueFinder(neighbors,count)
  127. x._2.foreach(r=>{
  128. valv1=r._1.toString
  129. valv2=r._2.toString
  130. if(neighbors.containsKey(v1)){
  131. neighbors.get(v1).add(v2)
  132. }else{
  133. valtemp=newutil.HashSet[String]()
  134. temp.add(v2)
  135. neighbors.put(v1,temp)
  136. }
  137. if(neighbors.containsKey(v2)){
  138. neighbors.get(v2).add(v1)
  139. }else{
  140. valtemp=newutil.HashSet[String]()
  141. temp.add(v1)
  142. neighbors.put(v2,temp)
  143. }
  144. })
  145. println("BKCliqueFinder---"+x._1+"---"+System.currentTimeMillis())
  146. finder.findMaxCliques().asScala
  147. }
  148. //jgrapht中的极大团算法
  149. deffind2(x:(String,List[(Long,Long)]),count:Int):Set[util.Set[String]]={
  150. println(x._1+"|s|"+x._2.size)
  151. println("BKCliqueFinder---"+x._1+"---"+System.currentTimeMillis())
  152. valto_clique=newSimpleGraph[String,DefaultEdge](classOf[DefaultEdge])
  153. x._2.foreach(r=>{
  154. valv1=r._1.toString
  155. valv2=r._2.toString
  156. to_clique.addVertex(v1)
  157. to_clique.addVertex(v2)
  158. to_clique.addEdge(v1,v2)
  159. })
  160. valfinder=newBronKerboschCliqueFinder(to_clique)
  161. vallist=finder.getAllMaximalCliques.asScala
  162. varresult=Set[util.Set[String]]()
  163. list.foreach(x=>{
  164. if(x.size()>=count)
  165. result=result+x
  166. })
  167. println("BKCliqueFinder---"+x._1+"---"+System.currentTimeMillis())
  168. result
  169. }
  170. }

//自己实现的极大团算法

  1. importjava.util.*;
  3. /**
  4. *[@author](https://my.oschina.net/arthor)mopspecial@gmail.com
  5. *[@date](https://my.oschina.net/u/2504391)2017/7/31
  6. */
  7. publicclassCliqueFinder{
  8. privateMap<String,Set<String>>neighbors;
  9. privateSet<String>nodes;
  10. privateSet<Set<String>>maxCliques=newHashSet<>();
  11. privateIntegerminSize;
  13. publicCliqueFinder(Map<String,Set<String>>neighbors,IntegerminSize){
  14. this.neighbors=neighbors;
  15. this.nodes=neighbors.keySet();
  16. this.minSize=minSize;
  17. }
  19. privatevoidbk3(Set<String>clique,List<String>candidates,List<String>excluded){
  20. if(candidates.isEmpty()&&excluded.isEmpty()){
  21. if(!clique.isEmpty()&&clique.size()>=minSize){
  22. maxCliques.add(clique);
  23. }
  24. return;
  25. }
  27. for(Strings:degeneracy_order(candidates)){
  28. List<String>new_candidates=newArrayList<>(candidates);
  29. new_candidates.retainAll(neighbors.get(s));
  31. List<String>new_excluded=newArrayList<>(excluded);
  32. new_excluded.retainAll(neighbors.get(s));
  33. Set<String>nextClique=newHashSet<>(clique);
  34. nextClique.add(s);
  35. bk2(nextClique,new_candidates,new_excluded);
  36. candidates.remove(s);
  37. excluded.add(s);
  38. }
  39. }
  41. privatevoidbk2(Set<String>clique,List<String>candidates,List<String>excluded){
  42. if(candidates.isEmpty()&&excluded.isEmpty()){
  43. if(!clique.isEmpty()&&clique.size()>=minSize){
  44. maxCliques.add(clique);
  45. }
  46. return;
  47. }
  48. Stringpivot=pick_random(candidates);
  49. if(pivot==null){
  50. pivot=pick_random(excluded);
  51. }
  52. List<String>tempc=newArrayList<>(candidates);
  53. tempc.removeAll(neighbors.get(pivot));
  55. for(Strings:tempc){
  56. List<String>new_candidates=newArrayList<>(candidates);
  57. new_candidates.retainAll(neighbors.get(s));
  59. List<String>new_excluded=newArrayList<>(excluded);
  60. new_excluded.retainAll(neighbors.get(s));
  61. Set<String>nextClique=newHashSet<>(clique);
  62. nextClique.add(s);
  63. bk2(nextClique,new_candidates,new_excluded);
  64. candidates.remove(s);
  65. excluded.add(s);
  66. }
  67. }
  69. privateList<String>degeneracy_order(List<String>innerNodes){
  70. List<String>result=newArrayList<>();
  71. Map<String,Integer>deg=newHashMap<>();
  72. for(Stringnode:innerNodes){
  73. deg.put(node,neighbors.get(node).size());
  74. }
  75. while(!deg.isEmpty()){
  76. Integermin=Collections.min(deg.values());
  77. StringminKey=null;
  78. for(Stringkey:deg.keySet()){
  79. if(deg.get(key).equals(min)){
  80. minKey=key;
  81. break;
  82. }
  83. }
  84. result.add(minKey);
  85. deg.remove(minKey);
  86. for(Stringk:neighbors.get(minKey)){
  87. if(deg.containsKey(k)){
  88. deg.put(k,deg.get(k)-1);
  89. }
  90. }
  92. }
  93. returnresult;
  94. }
  97. privateStringpick_random(List<String>random){
  98. if(random!=null&&!random.isEmpty()){
  99. returnrandom.get(0);
  100. }else{
  101. returnnull;
  102. }
  103. }
  105. publicSet<Set<String>>findMaxCliques(){
  106. this.bk3(newHashSet<>(),newArrayList<>(nodes),newArrayList<>());
  107. returnmaxCliques;
  108. }
  110. publicstaticvoidmain(String[]args){
  111. Map<String,Set<String>>neighbors=newHashMap<>();
  112. neighbors.put("0",newHashSet<>(Arrays.asList("1","2","3")));
  113. neighbors.put("1",newHashSet<>(Arrays.asList("0","2")));
  114. neighbors.put("2",newHashSet<>(Arrays.asList("0","1","3","6")));
  115. neighbors.put("3",newHashSet<>(Arrays.asList("0","2","4","5")));
  116. neighbors.put("4",newHashSet<>(Arrays.asList("3","5","6")));
  117. neighbors.put("5",newHashSet<>(Arrays.asList("3","4","6")));
  118. neighbors.put("6",newHashSet<>(Arrays.asList("2","4","5")));
  119. neighbors.put("7",newHashSet<>(Arrays.asList("6")));
  120. CliqueFinderfinder=newCliqueFinder(neighbors,3);
  121. finder.bk3(newHashSet<>(),newArrayList<>(neighbors.keySet()),newArrayList<>());
  122. System.out.println(finder.maxCliques);
  123. }
  124. }
转载请说明出处
知优网 » Spark Graphx 实现图中极大团挖掘, 伪并行化算法（spark graphx图算法案例）
站长资讯网友投稿帖

站长资讯网友投稿帖

分享到：

相关推荐

发表评论

您需要后才能发表评论