使用SparkStreaming和Kafka从HBase获取数据

Question

我正在尝试将SparkStreaming与HBase集成。我正在调用以下API来连接HBase：

HConnection hbaseConnection = HConnectionManager.createConnection(conf);
hBaseTable = hbaseConnection.getTable(hbaseTable);

因为我不能获得连接并广播连接，所以每次从HBase获取数据的API调用都非常昂贵。我尝试使用cdh5.5中的hbase-spark库来使用JavaHBaseContext (JavaHBaseContext hbaseContext = new JavaHBaseContext(jsc, conf))，但无法从maven导入该库。有没有人能够成功解决这个问题。

我正在尝试使用最新的API来连接Cloudera上的HBase和SparkStreaming。

这里提到的一些JIRA项目。

http://blog.cloudera.com/blog/2015/08/apache-spark-comes-to-apache-hbase-with-hbase-spark-module/

我使用JavaHBaseContext hbaseContext = new JavaHBaseContext(jssc.sparkContext(), conf);，然后调用bulk Get API hbaseContext.streamBulkGet(TableName.valueOf(tableName), 2, lines, new GetFunction2(), new ResultFunction());

但是这个批量API是在初始化期间调用的，而不是在每个流消息期间调用。还使用了：

hbaseContext.foreachPartition(jDStream,new VoidFunction<Tuple2<Iterator<String>, Connection>>() {
      public void call(Tuple2<Iterator<String>, Connection> t)throws Exception { ...}

API存在，但不知何故，它不能用于流消息。我也尝试过hbaseContext.streamMap(jdstream, new Function<Tuple2<Iterator<String>, Connection>, Iterator<String>>()，但它也不起作用。

我们有一个如何使用spark streaming API获取数据的示例吗？

Answer 1

您在哪里建立连接？如果连接代码仅在驱动程序上，请确保连接对象是可序列化的

我正在使用Cassandra，我所做的是我有一个scala对象，其中我有cassandra connection对象。这样，在驱动程序和工作程序上，至少有一个exeuctor作用域的连接对象