Spark Streaming with HBase with filtering logic

Question

나는 스파크 스트리밍과 hbase 연결 방법을 이해하려고 노력했지만 성공하지 못했습니다. 내가 뭘하려고하는지 스파크 스트림을 주어진 스트림을 처리하고 hbase 테이블에 결과를 저장합니다. 지금까지 내가 가지고있는 것입니다 :

import org.apache.spark.SparkConf 
import org.apache.spark.streaming.{Seconds, StreamingContext} 
import org.apache.spark.streaming.StreamingContext._ 
import org.apache.spark.storage.StorageLevel 
import org.apache.hadoop.hbase.HBaseConfiguration 
import org.apache.hadoop.hbase.client.{HBaseAdmin,HTable,Put,Get} 
import org.apache.hadoop.hbase.util.Bytes 

def blah(row: Array[String]) { 
    val hConf = new HBaseConfiguration() 
    val hTable = new HTable(hConf, "table") 
    val thePut = new Put(Bytes.toBytes(row(0))) 
    thePut.add(Bytes.toBytes("cf"), Bytes.toBytes(row(0)), Bytes.toBytes(row(0))) 
    hTable.put(thePut) 
} 

val ssc = new StreamingContext(sc, Seconds(1)) 
val lines = ssc.socketTextStream("localhost", 9999, StorageLevel.MEMORY_AND_DISK_SER) 
val words = lines.map(_.split(",")) 
val store = words.foreachRDD(rdd => rdd.foreach(blah)) 
ssc.start() 

나는 spark-shell에서 위의 코드를 현재 실행하고 있습니다. 내가 뭘 잘못하고 있는지 모르겠다.
나는 쉘에서 다음과 같은 오류가 발생합니다 :

14/09/03 16:21:03 ERROR scheduler.JobScheduler: Error running job streaming job 1409786463000 ms.0 

org.apache.spark.SparkException: Job aborted due to stage failure: Task not serializable: java.io.NotSerializableException: org.apache.spark.streaming.StreamingContext 

at org.apache.spark.scheduler.DAGScheduler.org$apache$spark$scheduler$DAGScheduler$$failJobAndIndependentStages(DAGScheduler.scala:1033) 

at org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1017) 

at org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1015) 

at scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59) 

at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:47) 

at org.apache.spark.scheduler.DAGScheduler.abortStage(DAGScheduler.scala:1015) 

at org.apache.spark.scheduler.DAGScheduler.org$apache$spark$scheduler$DAGScheduler$$submitMissingTasks(DAGScheduler.scala:770) 

at org.apache.spark.scheduler.DAGScheduler.org$apache$spark$scheduler$DAGScheduler$$submitStage(DAGScheduler.scala:713) 

at org.apache.spark.scheduler.DAGScheduler.handleJobSubmitted(DAGScheduler.scala:697) 

at org.apache.spark.scheduler.DAGSchedulerEventProcessActor$$anonfun$receive$2.applyOrElse(DAGScheduler.scala:1176) 

at akka.actor.ActorCell.receiveMessage(ActorCell.scala:498) 

at akka.actor.ActorCell.invoke(ActorCell.scala:456) 

at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:237) 

at akka.dispatch.Mailbox.run(Mailbox.scala:219) 

at akka.dispatch.ForkJoinExecutorConfigurator$AkkaForkJoinTask.exec(AbstractDispatcher.scala:386) 

at scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260) 

at scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339) 

at scala.concurrent.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979) 

at scala.concurrent.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107) 

는 또한 두 단지 경우, HBase를 테이블을 확인하고, 새로운 아무것도 거기에 기록되지 않습니다.

다른 터미널에서 nc -lk 9999를 실행하여 테스트를 위해 spark-shell에 데이터를 입력하고 있습니다.

Answer 1

spark 사용자 그룹의 도움을 받아 어떻게 작동하는지 파악할 수있었습니다. 나는 직렬화 객체 주위에 내 스트리밍, 매핑 및 foreach는 호출을 래핑하는 데 필요한 것 같습니다 :

import org.apache.spark.SparkConf 
import org.apache.spark.streaming.{Seconds, StreamingContext} 
import org.apache.spark.streaming.StreamingContext._ 
import org.apache.spark.storage.StorageLevel 
import org.apache.hadoop.hbase.HBaseConfiguration 
import org.apache.hadoop.hbase.client.{HBaseAdmin,HTable,Put,Get} 
import org.apache.hadoop.hbase.util.Bytes 

object Blaher { 
    def blah(row: Array[String]) { 
    val hConf = new HBaseConfiguration() 
    val hTable = new HTable(hConf, "table") 
    val thePut = new Put(Bytes.toBytes(row(0))) 
    thePut.add(Bytes.toBytes("cf"), Bytes.toBytes(row(0)), Bytes.toBytes(row(0))) 
    hTable.put(thePut) 
    } 
} 

object TheMain extends Serializable{ 
    def run() { 
    val ssc = new StreamingContext(sc, Seconds(1)) 
    val lines = ssc.socketTextStream("localhost", 9999, StorageLevel.MEMORY_AND_DISK_SER) 
    val words = lines.map(_.split(",")) 
    val store = words.foreachRDD(rdd => rdd.foreach(Blaher.blah)) 
    ssc.start() 
    } 
} 

TheMain.run() 

Answer 2

은 일반적인 안티 패턴 것 같다. 올바른 패턴을 보려면 http://spark.apache.org/docs/latest/streaming-programming-guide.html의 "foreachRDD 사용을위한 디자인 패턴"장을 참조하십시오.