sparkstreaming读取kafka的数据并写入到hbase

Spark Streaming可以通过Kafka Direct方式读取Kafka的数据，并将数据写入到HBase中。

具体步骤如下：

1. 创建Spark Streaming上下文

val conf = new SparkConf().setAppName("KafkaToHBase")
val ssc = new StreamingContext(conf, Seconds(5))

2. 创建Kafka Direct流

val kafkaParams = Map[String, String](
  "metadata.broker.list" -> "localhost:9092",
  "group.id" -> "test-group"
)

val topics = Set("test-topic")

val kafkaStream = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](
  ssc, kafkaParams, topics)

3. 解析Kafka数据

val messages = kafkaStream.map(_._2)

4. 将数据写入HBase

messages.foreachRDD(rdd => {
  rdd.foreachPartition(partition => {
    val hbaseConf = HBaseConfiguration.create()
    val connection = ConnectionFactory.createConnection(hbaseConf)
    val table = connection.getTable(TableName.valueOf("test-table"))

    partition.foreach(record => {
      val put = new Put(Bytes.toBytes(record))
      put.addColumn(Bytes.toBytes("cf"), Bytes.toBytes("col"), Bytes.toBytes(record))
      table.put(put)
    })

    table.close()
    connection.close()
  })
})

5. 启动Spark Streaming

ssc.start()
ssc.awaitTermination()

以上就是使用Spark Streaming读取Kafka的数据并写入到HBase的步骤。