Scala - Tweets 订阅 - Kafka 主题和摄取到 HBase

Question

我必须使用来自 Kafka 主题的推文并将其摄取到 HBase 中。以下是我编写的代码，但无法正常工作。

主要代码没有调用“convert”方法，因此没有记录被摄取到 HBase 表中。谁能帮帮我。

tweetskafkaStream.foreachRDD(rdd => {
  println("Inside For Each RDD" )
  rdd.foreachPartition( record => {
    println("Inside For Each Partition" )
    val data = record.map(r => (r._1, r._2)).map(convert)
    })
  })

def convert(t: (String, String)) = {
    println("in convert")
    //println("first param value ", t._1)
    //println("second param value ", t._2)

  val hConf = HBaseConfiguration.create()
  hConf.set(TableOutputFormat.OUTPUT_TABLE,hbaseTableName)
  hConf.set("hbase.zookeeper.quorum", "192.168.XXX.XXX:2181")
  hConf.set("hbase.master", "192.168.XXX.XXX:16000")
  hConf.set("hbase.rootdir","hdfs://192.168.XXX.XXX:9000/hbase")
  val today = Calendar.getInstance.getTime
  val printformat =  new SimpleDateFormat("yyyyMMddHHmmss")

  val id =  printformat.format(today)
  val p = new Put(Bytes.toBytes(id))

  p.add(Bytes.toBytes("data"), Bytes.toBytes("tweet_text"),(t._2).getBytes())
  (id, p)

  val mytable = new HTable(hConf,hbaseTableName)
  mytable.put(p)
}

我不想使用当前日期时间作为键 (t._1) 并因此在我的转换方法中构建它。

谢谢

巴拉

Answer 1

我将其改为 foreach，而不是 foreachPartition。这很好用。