我正在撰写一个项目，从Kafka接收数据并将其写入Hbase表中。因为我想知道记录的差异，所以需要先获取Hbase中具有相同行键的记录，然后对接收到的记录进行减法运算，最后将新记录保存到HBase表中。
一开始，我尝试使用
通过这种方式，我可以仅一次获取具有特定列族和列名的记录的值。所谓“仅一次”，是指每次启动我的spark流式应用程序时，这段代码将被执行并且我可以获取一个值，但它不会再执行了。由于我想每次从Kafka接收记录时都使用cf和列从HBase读取我的记录，所以这对我不起作用。
为解决这个问题，我将逻辑移动到foreachRDD()中，但不幸的是，sparkContext似乎不可序列化。我收到了类似“任务不可序列化”的错误。
最后，我发现还有另一种方法可以使用hbase.client HTable从hbase读取数据。因此，这是我的最终工作：
这对我来说现在运行良好，但我对这个过程有疑问：
我猜想，对于RDD的每个分区，都会创建一个连接到HBase的连接。我想知道是否可能扩展我的应用程序。比如说，如果我在1秒钟内有超过1000条记录，看起来我的Spark Streaming中将建立1000个连接。
这是从HBase读取数据的正确方法吗？在SparkStreaming中读取数据的最佳实践是什么？还是Spark Streaming不应该读取任何数据，而只是设计用于将流式数据写入数据库。
提前致谢。
一开始，我尝试使用
newAPIHadoop从Hbase获取数据。以下是我的尝试：val conf = HBaseConfiguration.create()
conf.set("zookeeper.znode.parent", "/hbase-secure")
conf.set(TableOutputFormat.OUTPUT_TABLE, tableName)
conf.set("hbase.zookeeper.quorum", zkQuorum)
conf.set("hbase.master", masterAddr)
conf.set("hbase.zookeeper.property.clientPort", portNum)
conf.set(TableInputFormat.INPUT_TABLE, tableName)
conf.set(TableInputFormat.SCAN_COLUMNS, cfName + ":" + colName)
val HbaseRDD = ssc.sparkContext.newAPIHadoopRDD(conf,
classOf[TableInputFormat],
classOf[org.apache.hadoop.hbase.io.ImmutableBytesWritable],
classOf[org.apache.hadoop.hbase.client.Result])
通过这种方式，我可以仅一次获取具有特定列族和列名的记录的值。所谓“仅一次”，是指每次启动我的spark流式应用程序时，这段代码将被执行并且我可以获取一个值，但它不会再执行了。由于我想每次从Kafka接收记录时都使用cf和列从HBase读取我的记录，所以这对我不起作用。
为解决这个问题，我将逻辑移动到foreachRDD()中，但不幸的是，sparkContext似乎不可序列化。我收到了类似“任务不可序列化”的错误。
最后，我发现还有另一种方法可以使用hbase.client HTable从hbase读取数据。因此，这是我的最终工作：
def transferToHBasePut(line: String): (ImmutableBytesWritable, Put) = {
val conf = HBaseConfiguration.create()
conf.set("zookeeper.znode.parent", "/hbase-secure")
conf.set("hbase.zookeeper.quorum", "xxxxxx")
conf.set("hbase.master", "xxxx")
conf.set("hbase.zookeeper.property.clientPort", "xxx")
conf.set(TableInputFormat.INPUT_TABLE, "xx")
conf.set(TableInputFormat.SCAN_COLUMNS, "xxxxx")
val testTable = new HTable(conf, "testTable")
val scan = new Scan
scan.addColumn("cf1".getBytes, "test".getBytes)
val rs = testTable.getScanner(scan)
var r = rs.next()
val res = new StringBuilder
while(r != null){
val tmp = new String(r.getValue("cf1".getBytes, "test".getBytes))
res.append(tmp)
r= rs.next()
}
val res = res.toString
//do the following manipulations and return object (ImmutableBytesWritable, Put)
..............................
.......................
}
在主方法中，我使用上述方法在foreachRDD中使用saveAsNewAPIHadoopDataset方法将其保存到HBase中。
streamData.foreachRDD(stream => stream.map (transferToHBasePut).saveAsNewAPIHadoopDataset(job.getConfiguration))
这对我来说现在运行良好，但我对这个过程有疑问：
我猜想，对于RDD的每个分区，都会创建一个连接到HBase的连接。我想知道是否可能扩展我的应用程序。比如说，如果我在1秒钟内有超过1000条记录，看起来我的Spark Streaming中将建立1000个连接。
这是从HBase读取数据的正确方法吗？在SparkStreaming中读取数据的最佳实践是什么？还是Spark Streaming不应该读取任何数据，而只是设计用于将流式数据写入数据库。
提前致谢。