一、自定義圖

import org.apache.spark.{SparkConf, SparkContext} import org.apache.spark.graphx.{Edge, VertexId} import org.apache.spark.rdd.RDD import org.apache.spark.graphx.Graphobject PropertiesGraph {def main(args: Array[String]): Unit = {val conf = new SparkConf().setAppName("graph").setMaster("local[*]")val sc = new SparkContext(conf)sc.setLogLevel("OFF")//定義點val users:RDD[(VertexId,(String,String))] = sc.parallelize(Array((3L,("s1","student")),(6L,("s2","teacher")),(5L,("s3","pro")),(2L,("s4","pro"))))//定義邊val relationships:RDD[Edge[String]] = sc.parallelize(Array(Edge(3L,6L,"cooperate"),Edge(5L,6L,"agvisor"),Edge(2L,5L,"colleague")))val defaultUser = ("Jack ma","defaultUser")val graph = Graph(users,relationships,defaultUser)}}

二、圖基本操作

import org.apache.spark.{SparkConf, SparkContext} import org.apache.spark.graphx.{Edge, VertexId} import org.apache.spark.rdd.RDD import org.apache.spark.graphx.Graphobject PropertiesGraph {def main(args: Array[String]): Unit = {val conf = new SparkConf().setAppName("graph").setMaster("local[*]")val sc = new SparkContext(conf)sc.setLogLevel("OFF")//定義點val users:RDD[(VertexId,(String,String))] = sc.parallelize(Array((3L,("s1","student")),(6L,("s2","teacher")),(5L,("s3","pro")),(2L,("s4","pro"))))//定義邊val relationships:RDD[Edge[String]] = sc.parallelize(Array(Edge(3L,6L,"cooperate"),Edge(5L,6L,"advisor"),Edge(2L,5L,"colleague")))val defaultUser = ("Jack ma","defaultUser")val graph = Graph(users,relationships,defaultUser)/********************** 圖操作 ***************************///找到途中屬性是student的點graph.vertices.filter{case (id,(name,work))=>work=="student"}.collect.foreach{case (id,(name,work))=>println(s"$name is $work")}//找到途中屬性為advisor的邊graph.edges.filter(x => x.attr=="advisor").collect().foreach(x=>println(s"${x.srcId} to ${x.dstId} 屬性為 ${x.attr}"))//出度和入度操作def max(a:(VertexId,Int),b:(VertexId,Int)):(VertexId,Int) = {if(a._2 > b._2) a else b}println("最大的出度" + graph.outDegrees.reduce(max))println("最大的入度" + graph.inDegrees.reduce(max))println("最大的度數" + graph.degrees.reduce(max))// 給每一個點的職業屬性上加一個字符串graph.mapVertices{case (id,(name,work))=>(id,(name,work+"_spark"))}.vertices.collect.foreach(println)// triplet:a關系b a -> b// 給圖中每個元組的Edge的屬性值設置為源的屬性值+邊的屬性值+目標點的屬性值graph.mapTriplets(x=>x.srcAttr._2 + "+" + x.attr + "+" + x.dstAttr._2).edges.collect().foreach(println)// 兩者之間的關系graph.triplets.map(x=>x.srcAttr._1 + " is the " + x.attr + " of " + x.dstAttr._1).foreach(println)// 刪除節點，構建子圖val validGraphx = graph.subgraph(vpred = (id,attr)=> attr._2 != "pro")validGraphx.vertices.foreach(println)validGraphx.triplets.map(x=>x.srcAttr._1 + " is the " + x.attr + " of " + x.dstAttr._1).foreach(println)// 構建職業是professor的子圖val subGraph = graph.subgraph(vpred = (id, attr)=>attr._2=="pro")subGraph.vertices.collect().foreach(x=>println(s"${x._2._1} is ${x._2._2}"))}}

三、Spark加載neo4j

import org.apache.spark.SparkConf import org.apache.spark.api.java.JavaSparkContext import org.apache.spark.ml.tree.InternalNode import org.neo4j.spark.Neo4jobject SparkGraphxConnector {def main(args: Array[String]): Unit = {val conf = new SparkConf().setAppName("neo4j").setMaster("local[*]").set("spark.neo4j.bolt.url","bolt://ip").set("spark.neo4j.bolt.user","user_name").set("spark.neo4j.bolt.password","password")val sc = new JavaSparkContext(conf)sc.setLogLevel("OFF")val neo4j = Neo4j(sc)val rdd = neo4j.cypher("match (n:Person) return n").loadRowRddval personRDD = rdd.map(row=>{val map = row.get(0).asInstanceOf[InternalNode]new Person(map.get("home").asString(),map.get("name").asString(),map.get("personID").asString())})personRDD.foreach(println)// 圖查詢val graphQuery = "match (n:Person)-[r]-(a:Person) return id(p) as source,id(a) as target,type(r) as value"val graph:Graph[String,String] = neo4j.rels(graphQuery).loadGraphgraph.edges.foreach(println(_))}case class Person(val home:String,val name:String,val personID:String)}

?

總結

以上是生活随笔為你收集整理的知识图谱学习笔记－图操作的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。