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

길고도 짧은 이야기, 당신은 할 수 없습니다 broadcastGraph를 그것이 RDD (그 중 실제로 부부)가 포함되어 있기 때문이다. 그리고 RDD의 map 기능에있는 Graph은 RDDs으로 구성되어 있기 때문에 사용할 수 없습니다.

내가 말했듯이, 당신이 그 중 하나를 할 수없는 긴 이야기입니다 - 그리고 그들은 정말로 같은 동전의 두면입니다. 그것은 당신이 어느쪽으로도 반대하는 것과 같은 이슈입니다.

스파크는 마스터/슬레이브 개념을 기반으로합니다. 마스터의 메모리 공간에 있으며 RDDs과 관련된 메타 조치가 정의되어 있습니다. 하지만 코드 ... (map(...))은 슬레이브 (executor라고 함)에서 실행됩니다. 다른 RDD을 어떤 식 으로든 참조 할 때지도 작성자는지도 작성자가 실행할 수 없습니다. RDD 참조는 마스터에만 존재할 수 있기 때문에 broadcast은 절대 도움이되지 않습니다.

대신 무엇을 할 수 있습니까? 당신은 두 가지 옵션이 있습니다

1. 당신이 collect()을 사용하여 필요한 데이터를 수집, 그리고 하나 데이터 또는 당신의 map 코드의 내부에 그것을 참조 broadcast. collect()은 모든 데이터를 마스터로 가져 오지만 가장 중요한 문제는 RDD 참조를 사용하지 않고 데이터를 참조 할 수 있으므로 수집 된 데이터를 broadcast을 사용하거나 map(...) 코드를 참조하십시오 (스파크는 집행자에게 데이터 사본을 발송합니다). 어떤 작업이든 또는 둘 중 어느 작업이든 데이터 크기, 속도 기대치 등에 따라 달라집니다.
2. RDD.join() 또는 RDD.cogroup()을 사용하면 두 개의 Graphs을 동시에 사용할 수 있습니다.

두 가지 모두 고차원 구조 인 GraphX ​​Graph에서 작업한다는 점에서 복잡합니다. 별도의 Graph.vertices과 Graph.edgesRDDs을 별도로 작업해야하며 collect() 또는 join()을 작성한 다음 RDDs을 함께 바느질하여 Graph의 마지막을 다시 빌드하십시오.