sparkstreaming读Kafka写hbase

Spark Streaming可以很方便地读取Kafka中的数据，并将其写入到HBase中。这里是一些基本的步骤：

1. 首先，你需要在你的Spark Streaming项目中添加Kafka和HBase的依赖。

2. 创建一个Spark Streaming的StreamingContext对象，并设置批处理间隔。

3. 创建一个Kafka输入流，指定Kafka的相关参数，如Kafka集群地址、消费者组、Topic等。

4. 从输入流中获取数据，进行必要的转换和处理，然后将结果写入到HBase中。写入HBase可以使用HBase的Java API或者使用Spark对HBase的支持。

下面是一个简单的示例代码，假设你已经完成了步骤1和步骤2：

import org.apache.spark.streaming._
import org.apache.spark.streaming.kafka._
import org.apache.hadoop.hbase.client.{HBaseAdmin,HTable,Put}
import org.apache.hadoop.hbase.{HBaseConfiguration, HTableDescriptor, TableName}
import org.apache.hadoop.hbase.util.Bytes

val ssc = new StreamingContext(sparkConf, Seconds(5))

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

val messages = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc, kafkaParams, topicsSet)

messages.foreachRDD { rdd =>
  rdd.foreach { msg =>
    val hconf = HBaseConfiguration.create()
    val hTable = new HTable(hconf, "table_name")

    val put = new Put(Bytes.toBytes(msg._1))
    put.add(Bytes.toBytes("cf"), Bytes.toBytes("col"), Bytes.toBytes(msg._2))
    hTable.put(put)
  }
}

ssc.start()
ssc.awaitTermination()

这个示例中，我们使用了KafkaUtils.createDirectStream方法创建了一个Kafka输入流，然后使用foreachRDD方法对每个RDD进行处理。在处理过程中，我们创建了一个HBase表实例，并将数据写入到表中。这里只是一个简单的示例，实际上你需要根据你的数据格式和业务逻辑进行相应的处理和转换。