foreach 내부의 Spark Streaming 필터 조건 - NullPointerException

Question

HBase 테이블에 마스터 정보가있는 스트림 데이터가 있습니다. 모든 행에 대해 HBase 마스터 테이블을 조회하고 프로필 정보를 얻어야합니다. 내 코드는 다음과 같습니다.

val con    = new setContext(hadoopHome,sparkMaster) 
val l_sparkcontext = con.getSparkContext 
val l_hivecontext = con.getHiveContext 

val topicname  = "events" 
val ssc    = new StreamingContext(l_sparkcontext, Seconds(30)) 
val eventsStream = KafkaUtils.createStream(ssc,"xxx.xxx.142.xxx:2181","receive_rest_events",Map(topicname.toString -> 10)) 
println("Kafka Stream for receiving Events..") 

val profile_data = l_hivecontext.sql("select gender, income, age, riid from hbase_customer_profile") 
profile_data.foreach(println) 
val tabBC = l_sparkcontext.broadcast(profile_data) 

eventsStream.foreachRDD(rdd => { 
    rdd.foreach(record => { 
    val subs_profile_rows = tabBC.value 
    val Rows = record._2.split(rowDelim) 
    Rows.foreach(row => { 
     val values = row.split(colDelim) 
     val riid = values(1).toInt 
     val cond = "riid = " + riid 
     println("Condition : ", cond) 
     val enriched_events = subs_profile_rows.filter(cond) 
    }) // End of Rows 
    }) // End of RDD 
}) // End of Events Stream 

불행히도 필자는 항상 필터에서 NPE를 누르십시오. 근로자 노드에서 값을 방송하기 위해 여기 몇 가지 질문과 답변을 따랐지만 아무것도 도움이되지 못했습니다. 누군가 제발 도와 줄 수 있어요.

관련

발라

는

Answer 1

귀하의 상황에 맞는 사용은 당신이 시도하는 두 개의 별도의 컨텍스트 (하나의 불꽃, 스파크 스트리밍을) 만들고있는 것처럼 약간 비린내가 ... 나에게 보이는 모습 이러한 컨텍스트간에 브로드 캐스트 변수를 공유합니다 (작동하지 않음).

우리는 비슷한 코드를 작성했습니다. 관심있는 경우에 대비하여 Splice Machine (오픈 소스)에서 어떻게했는지 보여주는 비디오가 있습니다. 나는 그 코드를 찾거나 다른 사람이 그 코드를 게시하도록 노력할 것이다.

http://community.splicemachine.com/splice-machine-tutorial-video-configuring-kafka-feed-splice-machine-part/

http://community.splicemachine.com/splice-machine-tutorial-video-configuring-kafka-feed-splice-machine-ii/

행운

.