spark-structstreaming-结果数据存入hbase

前言

本节描述通过spark-structstreaming将结果结果数据存入hbase

正文

object testWriteResultToHbase{
def main(args:Array[String]){

val kafkaservers=args(0)
val topic=args(1)
val zookeeperservers=args(2)
val tablename=args(3)
val spark=SparkSession
.builder
.appName("testWriteResultToHbase")
.master("local")
.getOrCreate()

import spark.implicits._
val lines = spark
.readStream
.format("kafka")
.option("kafka.bootstrap.servers",kafkaservers)
.option("subscribe", topic)
.load()
.selectExpr("cast(topic as String) ","cast(key as String)","CAST(value AS STRING)")
.as[(String,String,String)]
lines.createTempView("Originalkafka")
import spark.sql
val count=sql("select value from Originalkafka ")
val query =count.writeStream
.outputMode("append")
.foreach(new ForeachWriter[Row]{

var connection: Connection= null
def open(partitionId:Long,version:Long):Boolean={
true
}
def process(record:Row):Unit={
val conf = new HBaseConfiguration()
conf.set("hbase.zookeeper.quorum",zookeeperservers)
conf.set("hbase.zookeeper.property.clientPort", "2181")
conf.set("zookeeper.znode.parent","/hbase-unsecure")
val table = new HTable(conf, tablename)
val theput= new Put(Bytes.toBytes(record.mkString))
theput.add(Bytes.toBytes("cf1"),Bytes.toBytes("age"),Bytes.toBytes("30"))
table.put(theput)
}
def close(errorOrNull:Throwable):Unit={
}
})
.queryName("test")
.format("foreach")
.start()
query.awaitTermination()
}

}

注意：写入hbase的时候，没有创建连接的客户端

ConnectionFactory.createConnection(conf)

所以就没有必要在

open(partitionId:Long,version:Long)

和

close(errorOrNull:Throwable)

打开连接和关闭连接。

但是本人认为还是有必要的。空了研究一下hbase 客户端源码