我有一个用scala编写的apachespark应用程序,它试图从hbase读取数据并对其进行处理。
我遇到过这样做的方法,以及如何使用spark流来实现这一点
所以我写了以下代码:
def main(args: Array[String]): Unit = {
val configuration = HBaseConfiguration.create()
configuration.set(TableInputFormat.INPUT_TABLE, "urls")
configuration.set(TableInputFormat.SCAN_COLUMNS, "values:words")
val hbaseRdd = sc.newAPIHadoopRDD(configuration,
classOf[TableInputFormat],
classOf[ImmutableBytesWritable],
classOf[Result]
)
val data = hbaseRdd.map(entry => {
val result = entry._2
Bytes.toString(result.getRow)
})
data.foreach(println)
}
我的hbase表是这样创建的: create 'urls', {NAME => 'values', VERSIONS => 5}
我得到的是:
16/03/10 17:10:17 ERROR TableInputFormat: java.io.IOException: java.lang.reflect.InvocationTargetException
at org.apache.hadoop.hbase.client.ConnectionFactory.createConnection(ConnectionFactory.java:240)
at org.apache.hadoop.hbase.client.ConnectionFactory.createConnection(ConnectionFactory.java:218)
at org.apache.hadoop.hbase.client.ConnectionFactory.createConnection(ConnectionFactory.java:119)
at org.apache.hadoop.hbase.mapreduce.TableInputFormat.initialize(TableInputFormat.java:183)
at org.apache.hadoop.hbase.mapreduce.TableInputFormatBase.getSplits(TableInputFormatBase.java:241)
at org.apache.hadoop.hbase.mapreduce.TableInputFormat.getSplits(TableInputFormat.java:237)
在阅读了有关此异常的内容之后,我可能应该将此添加为堆栈跟踪的一部分:
Caused by: java.lang.reflect.InvocationTargetException
at sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method)
at sun.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:62)
at sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)
at java.lang.reflect.Constructor.newInstance(Constructor.java:423)
at org.apache.hadoop.hbase.client.ConnectionFactory.createConnection(ConnectionFactory.java:238)
... 34 more
Caused by: java.lang.ClassCastException: org.apache.hadoop.hbase.ipc.RpcClientImpl cannot be cast to org.apache.hadoop.hbase.ipc.RpcClient
at org.apache.hadoop.hbase.ipc.RpcClientFactory.createClient(RpcClientFactory.java:64)
at org.apache.hadoop.hbase.ipc.RpcClientFactory.createClient(RpcClientFactory.java:48)
at org.apache.hadoop.hbase.client.ConnectionManager$HConnectionImplementation.<init>(ConnectionManager.java:637)
... 39 more
我的问题是:
有人能介绍一下使用spark从hbase检索数据的一些基本方法吗。最好是比我展示的链接更新得多的东西
如果我在代码中做错了什么,如果你能告诉我是什么,我会很感激的
如果我能以某种方式将数据作为一个Dataframe来读取,那就更好了
我使用的是spark 1.6.0和hbase 1.2.0
提前谢谢
1条答案
按热度按时间xurqigkl1#
很显然,这是一个意外的依赖性问题(当它没有任何意义时总是如此)。
以下是我为解决这个问题而采取的步骤(希望它们能帮助未来的开发人员):
我用完全相同的代码创建了一个干净的项目。这没有任何问题,这立即让我怀疑这是某种依赖性问题
为了确保这一点,我将hbase依赖项放在依赖项的顶部。这创建了一个与spark和安全性相关的不同的异常,更具体地说是:javax.servlet.filterregistration
然后我遇到了这个有用的解决方案,为我解决了这个问题。我不得不从pom中排除所有javax和mortbay jetty。这解决了我所有的问题
就这样:)