使用graphx構建的圖形沒有得到正確的廣播

Question

我使用graphx創建了一個圖形，現在我需要從原始圖形中提取子圖形。 users_graph是具有索引給用戶的子圖的RDD。問題是這些子圖沒有被計算出來。當我試圖對這些子圖進行操作時，我得到一個java.lang.NullPointerException異常。

class VertexProperty(val id:Long) extends Serializable 
case class User(val userId:Long, var offset:Int, val userCode:String, val Name:String, val Surname:String, val organizational_unit:String, val UME:String, val person_type:String, val SOD_HIGH:String, val SOD_MEDIUM:String, val SOD_LOW:String, val Under_mitigated:String) extends VertexProperty(userId) 
case class Account(val accountId:Long, var offset:Int, val userCode:String, val userId:String, val account_creation_date:String, var disabled:String, var forcechangepwd:String, var pwdlife:String, var numberloginerror:String, var lastchangepwd:String, var lastlogin:String, var lastwronglogin:String, var state:String, var expire:String, var last_cert_time:String, var creation_date:String, var creation_user:String,var challenge_counter:String, var challenge_failed_attempt:String) extends VertexProperty(accountId) //Check if userCode is actually the code in this example. 
case class Application(var applicationId:Long, var offset:Int, var Name:String, var Description:String, var Target:String, var Owner:String, var Ownercode:String, var Creation_date:String, var Creation_user:String) extends VertexProperty(applicationId) 
case class Entitlement(val entitlementId:Long, var offset:Int, val Name:String, var Code:String, var Description:String, var Type:String, var Application:String, var Administrative:String, var Parent_ID:String, var Owner_code:String, var Scope_type:String, var Business_name:String, var Business_policy:String, var SOD_high:String, var SOD_medium:String, var SOD_low:String) extends VertexProperty(entitlementId) 

/* 
Some code for computing vertexRDD and edges 
*/  

val graph: Graph[VertexProperty,String] = Graph(vertexRDD, edges, new VertexProperty(-1)) 
val triplets = graph.triplets 
val temp = triplets.map(t => t.attr) 
val distinct_users = temp.distinct.filter(t => t != "NULL") 

var bcast_graph = sc.broadcast(graph) 
val users_graph = distinct_users.map(du => du -> bcast_graph.value.subgraph(epred = t => t.attr == du)) 

Answer 1

長話短說，你不能broadcast一個Graph，因爲它包含一個RDD（實際上他們夫婦）。並且您不能在RDD的map函數中使用Graph函數，因爲它由RDDs組成。

就像我說過的，爲什麼你不能做那兩個 - 而且他們真的是同一枚硬幣的兩面。這是同樣的問題，你反對任何方式。

Spark建立在主/從概念之上。它在主內存空間中，其中定義了RDDs及其關聯的元操作。但是代碼...在map(...)之內 - 在從站（稱爲執行程序）上執行。當您以任何方式引用不同的RDD時，您的地圖代碼無法在執行程序內運行 - 並且broadcast永遠不會提供幫助，因爲RDD引用只能存在於主文件中。

你能做什麼呢？你有兩個選擇：

1. 收集所需使用collect()的數據，要麼broadcast的數據或只是指它的你map代碼中。 collect()所有數據駛入主，但最重要的是你的問題，這意味着你現在可以參考的數據，而無需使用RDD引用，這樣你就可以出貨所收集的數據與您的執行人 - 無論是使用broadcast或只是通過在您的map(...)代碼中引用它們（Spark將數據的副本發送給您的執行者）。這些作品中的哪一個或哪一個將取決於您的數據大小，速度預期等。
2. 使用RDD.join()或RDD.cogroup()可以同時處理兩個Graphs。

由於您正在研究高階結構 - GraphX Graph，所以這兩者都很複雜。您將不得不分別在單獨的Graph.vertices和Graph.edgesRDDs上工作，執行您的collect()或join()，然後通過將適當的RDDs縫合在一起重新構建最終的Graph。