Spark-Streaming-Filter-Bedingung innerhalb foreach - NullPointerException

Question

Wir haben eine Stream-Daten, für die ich einige Stamminformationen in einer HBase-Tabelle haben. Für jede Zeile muss ich nach der HBase-Haupttabelle suchen und einige Profilinformationen erhalten. Mein Code ist etwas wie das

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 

Leider traf ich immer auf NPE auf dem Filter. Ich hatte hier einige Fragen und Antworten verfolgt, um Werte über Arbeiterknoten zu übertragen, aber nichts hilft. Kann mir bitte jemand helfen.

Grüße

Bala

Answer 1

Ihre Kontextverwendung sieht ein bisschen faul ... Für mich ist es wie du (ein Funke, eine für Funken Streaming) sind die Schaffung von zwei getrennten Kontexten aussieht und dann zu versuchen, teilen Sie eine Broadcast-Variable zwischen diesen Kontexten (das wird nicht funktionieren).

Wir haben einen Code, den wir darüber geschrieben haben, der ähnlich war. Hier sind die Videos, die zeigen, wie wir es in Splice Machine (Open Source) gemacht haben, falls Sie interessiert sind. Ich werde versuchen, den Code zu finden oder ihn von jemand anderem posten lassen.

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/

Viel Glück.