星火RDD寫入HBase的

Question

我可以使用下面的代碼讀卡夫卡消息：

val ssc = new StreamingContext(sc, Seconds(50)) 
val topicmap = Map("test" -> 1) 
val lines = KafkaUtils.createStream(ssc,"127.0.0.1:2181", "test-consumer-group",topicmap) 

能否請你幫我，我怎麼能寫卡夫卡讀取到HBase的每封郵件？這是我想寫的，但沒有成功。

lines.foreachRDD(rdd => { 
    rdd.foreach(record => { 
    val i = +1 
    val hConf = new HBaseConfiguration() 
    val hTable = new HTable(hConf, "test") 
    val thePut = new Put(Bytes.toBytes(i)) 
    thePut.add(Bytes.toBytes("cf"), Bytes.toBytes("a"), Bytes.toBytes(record)) 
    }) 
}) 

Answer 1

那麼，你實際上並沒有執行Put，你是Mereley創建一個Put請求並向它添加數據。你缺少的是一個

hTable.put(thePut);