java - 使用 Spark 和 JAVA 从 HBase 读取数据

Question

我想使用 JAVA 通过 Spark 访问 HBase。除了 this 之外，我还没有找到任何例子一。答案中写着，

You can also write this in Java

我从 How to read from hbase using spark 复制了这段代码:

import org.apache.hadoop.hbase.client.{HBaseAdmin, Result}
import org.apache.hadoop.hbase.{ HBaseConfiguration, HTableDescriptor }
import org.apache.hadoop.hbase.mapreduce.TableInputFormat
import org.apache.hadoop.hbase.io.ImmutableBytesWritable

import org.apache.spark._

object HBaseRead {
  def main(args: Array[String]) {
    val sparkConf = new SparkConf().setAppName("HBaseRead").setMaster("local[2]")
    val sc = new SparkContext(sparkConf)
    val conf = HBaseConfiguration.create()
    val tableName = "table1"

    System.setProperty("user.name", "hdfs")
    System.setProperty("HADOOP_USER_NAME", "hdfs")
    conf.set("hbase.master", "localhost:60000")
    conf.setInt("timeout", 120000)
    conf.set("hbase.zookeeper.quorum", "localhost")
    conf.set("zookeeper.znode.parent", "/hbase-unsecure")
    conf.set(TableInputFormat.INPUT_TABLE, tableName)

    val admin = new HBaseAdmin(conf)
    if (!admin.isTableAvailable(tableName)) {
      val tableDesc = new HTableDescriptor(tableName)
      admin.createTable(tableDesc)
    }

    val hBaseRDD = sc.newAPIHadoopRDD(conf, classOf[TableInputFormat], classOf[ImmutableBytesWritable], classOf[Result])
    println("Number of Records found : " + hBaseRDD.count())
    sc.stop()
  }
}

谁能给我一些提示，告诉我如何找到正确的依赖项、对象和东西？

看起来 HBaseConfiguration 在 hbase-client 中，但我实际上卡在了 TableInputFormat.INPUT_TABLE 上。这不应该处于相同的依赖关系中吗？

有没有更好的方法用spark访问hbase？

Answer 1

TableInputFormat 类位于 hbase-server.jar 中，您需要在 pom.xml 中添加该依赖项。请查看HBase and non-existent TableInputFormat在 Spark 用户列表中。

<dependency>
    <groupId>org.apache.hbase</groupId>
    <artifactId>hbase-server</artifactId>
    <version>1.3.0</version>
</dependency>

下面是使用 Spark 从 Hbase 读取的示例代码。

public static void main(String[] args) throws Exception {
    SparkConf sparkConf = new SparkConf().setAppName("HBaseRead").setMaster("local[*]");
    JavaSparkContext jsc = new JavaSparkContext(sparkConf);
    Configuration hbaseConf = HBaseConfiguration.create();
    hbaseConf.set(TableInputFormat.INPUT_TABLE, "my_table");
    JavaPairRDD<ImmutableBytesWritable, Result> javaPairRdd = jsc.newAPIHadoopRDD(hbaseConf, TableInputFormat.class,ImmutableBytesWritable.class, Result.class);
    jsc.stop();
  }
}