使用Scala插入Hbase Issue的Spark Streaming

Question

我正在尝试从Kafka消息中读取记录并将其放入Hbase中。尽管scala脚本运行时没有任何问题，但插入操作并没有发生。请帮帮我。

输入：rowkey1,1rowkey2,2

这是我正在使用的代码：

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

object TheMain extends Serializable{
  def run() {
    val ssc = new StreamingContext(sc, Seconds(1)) 
    val topicmap = Map("test" -> 1)
    val lines = KafkaUtils.createStream(ssc,"127.0.0.1:2181", "test-consumer-group",topicmap).map(_._2)
    val words = lines.map(line => line.split(",")).map(line => (line(0),line(1)))
    val store = words.foreachRDD(rdd => rdd.foreach(Blaher.blah)) 
    ssc.start()
  } 
}
TheMain.run()

Answer 1

从HTable的flushCommits()方法的API文档："执行所有缓冲的Put操作"。您应该在blah()方法的末尾调用它——看起来它们当前正在缓冲中，但从未在某个随机时间执行或执行过。