十年網(wǎng)站開發(fā)經(jīng)驗(yàn) + 多家企業(yè)客戶 + 靠譜的建站團(tuán)隊(duì)
量身定制 + 運(yùn)營維護(hù)+專業(yè)推廣+無憂售后,網(wǎng)站問題一站解決
Spark圖處理GraphX學(xué)習(xí)筆記!

10年積累的成都網(wǎng)站設(shè)計(jì)、網(wǎng)站制作經(jīng)驗(yàn),可以快速應(yīng)對客戶對網(wǎng)站的新想法和需求。提供各種問題對應(yīng)的解決方案。讓選擇我們的客戶得到更好、更有力的網(wǎng)絡(luò)服務(wù)。我雖然不認(rèn)識你,你也不認(rèn)識我。但先網(wǎng)站設(shè)計(jì)后付款的網(wǎng)站建設(shè)流程,更有賀蘭免費(fèi)網(wǎng)站建設(shè)讓你可以放心的選擇與我們合作。
一、什么是GraphX?
Graphx利用了Spark這樣了一個并行處理框架來實(shí)現(xiàn)了圖上的一些可并行化執(zhí)行的算法。
算法是否能夠并行化與Spark本身無關(guān)
算法并行化與否的本身,需要通過數(shù)學(xué)來證明
已經(jīng)證明的可并行化算法,利用Spark來實(shí)現(xiàn)會是一個錯的選擇,因?yàn)镚raphx支持pregel的圖計(jì)算模型
二、Graphx包含哪些組件和基本框架?
graph中重要的成員變量分別為
vertices
edges
triplets
為什么要引入triplets呢,主要是和Pregel這個計(jì)算模型相關(guān),在triplets中,同時記錄著edge和vertex. 具體代碼就不羅列了。
函數(shù)分成幾大類
對所有頂點(diǎn)或邊的操作,但不改變圖結(jié)構(gòu)本身,如mapEdges, mapVertices
子圖,類似于集合操作中的filter subGraph
圖的分割,即paritition操作,這個對于Spark計(jì)算來說,很關(guān)鍵,正是因?yàn)橛辛瞬煌腜artition,才有了并行處理的可能, 不同的PartitionStrategy,其收益不同。最容易想到的就是利用Hash來將整個圖分成多個區(qū)域。
outerJoinVertices 頂點(diǎn)的外連接操作
圖的常用算法是集中抽象到GraphOps這個類中,在Graph里作了隱式轉(zhuǎn)換,將Graph轉(zhuǎn)換為GraphOps,具體有下列12個算子:
collectNeighborIds
collectNeighbors
collectEdges
joinVertices
filter
pickRandomVertex
pregel
pageRank
staticPageRank
connectedComponents
triangleCount
stronglyConnectedComponents
RDD是Spark體系的核心,那么Graphx中引入了哪些新的RDD呢,有倆,分別為
VertexRDD
EdgeRDD
較之EdgeRdd,VertexRDD更為重要,其上的操作也很多,主要集中于Vertex之上屬性的合并,說到合并就不得不扯到關(guān)系代數(shù)和集合論,所以在VertexRdd中能看到許多類似于sql中的術(shù)語,如
leftJoin
innerJoin
在進(jìn)行數(shù)學(xué)計(jì)算的時候,圖用線性代數(shù)中的矩陣來表示,那么如何進(jìn)行存儲呢?
學(xué)數(shù)據(jù)結(jié)構(gòu)的時候,老師肯定說過好多的辦法,不再啰嗦了。
不過在大數(shù)據(jù)的環(huán)境下,如果圖很巨大,表示頂點(diǎn)和邊的數(shù)據(jù)不足以放在一個文件中怎么辦? 用HDFS
加載的時候,一臺機(jī)器的內(nèi)存不足以容下怎么辦? 延遲加載,在真正需要數(shù)據(jù)時,將數(shù)據(jù)分發(fā)到不同機(jī)器中,采用級聯(lián)方式。
一般來說,我們會將所有與頂點(diǎn)相關(guān)的內(nèi)容保存在一個文件中vertexFile,所有與邊相關(guān)的信息保存在另一個文件中edgeFile。
生成某一個具體的圖時,用edge就可以表示圖中頂點(diǎn)的關(guān)聯(lián)關(guān)系,同時圖的結(jié)構(gòu)也表示出來了。
下面是Spark官方示例,用2個Array構(gòu)造了一個Graph。
val users: RDD[(VertexId, (String, String))] =
sc.parallelize(Array((3L, ("rxin", "student")), (7L, ("jgonzal", "postdoc")),
(5L, ("franklin", "prof")), (2L, ("istoica", "prof"))))
// Create an RDD for edges
val relationships: RDD[Edge[String]] =
sc.parallelize(Array(Edge(3L, 7L, "collab"), Edge(5L, 3L, "advisor"),
Edge(2L, 5L, "colleague"), Edge(5L, 7L, "pi")))
// Define a default user in case there are relationship with missing user
val defaultUser = ("John Doe", "Missing")
// Build the initial Graph
val graph = Graph(users, relationships, defaultUser)
graphLoader是graphx中專門用于圖的加載和生成,最重要的函數(shù)就是edgeListFile。
//以頂點(diǎn)劃分,分成4個分區(qū)
val graph = GraphLoader.edgeListFile(sc,"hdfs://192.168.0.10:9000/input/graph/web-Google.txt",minEdgePartitions = 4)
五、GraphX應(yīng)用舉例
一行代碼:
val rank = graph.pageRank(0.01).vertices
用RDD實(shí)現(xiàn):
// Connect to the Spark clusterval
sc = new SparkContext("spark://master.amplab.org", "research")
// Load my user data and parse into tuples of user id and attribute list
val users = (sc.textFile("graphx/data/users.txt")
.map(line => line.split(","))
.map( parts => (parts.head.toLong, parts.tail) ))
// Parse the edge data which is already in userId -> userId format
val followerGraph = GraphLoader.edgeListFile(sc, "graphx/data/followers.txt")
// Attach the user attributes
val graph = followerGraph.outerJoinVertices(users) {
case (uid, deg, Some(attrList)) => attrList
// Some users may not have attributes so we set them as empty
case (uid, deg, None) => Array.empty[String]
}
// Restrict the graph to users with usernames and names
val subgraph = graph.subgraph(vpred = (vid, attr) => attr.size == 2)
// Compute the PageRank
// Get the attributes of the top pagerank users
val userInfoWithPageRank = subgraph.outerJoinVertices(pagerankGraph.vertices) {
case (uid, attrList, Some(pr)) => (pr, attrList.toList)
case (uid, attrList, None) => (0.0, attrList.toList)
}
println(userInfoWithPageRank.vertices.top(5)(Ordering.by(_._2._1)).mkString("\n"))