Spark访问HBase的Eclipse代码实现

Hbase是一个列式数据库，从其本质上来看，可以当做是一个数据源，而Spark本身又可以进行Hbase的连接，访问数据并进行查询。

为了跟之前的程序对接，可以采用spark +hbase来实现数据的迁移和处理分析。因此小做了个实验测试一下。

(1) 建立scala project，导入hbase下的相关lib,当然这里面所需要的lib不多。只需要几个hbase开头的jar包即可，同时去掉一些结尾为.test.jar的包。

(2) 在Hbase中临时建个表，并输入条数据。如图所示。

(3) 在spark中利用原始的hbasetest.scala进行测试。     var tablename = "scores"     // Other options for configuring scan behavior are available. More information available at     // http://hbase.apache.org/apidocs/org/apache/hadoop/hbase/mapreduce/TableInputFormat.html     conf.set(TableInputFormat.INPUT_TABLE, tablename)     // Initialize hBase table if necessary     val admin = new HBaseAdmin(conf)     if (!admin.isTableAvailable(tablename)) {       val tableDesc = new HTableDescriptor(TableName.valueOf(args(0)))       admin.createTable(tableDesc)     }    println("start ")     val hBaseRDD = sc.newAPIHadoopRDD(conf, classOf[TableInputFormat],       classOf[org.apache.hadoop.hbase.io.ImmutableBytesWritable],       classOf[org.apache.hadoop.hbase.client.Result])    println(tablename + "表的总行数为 " +hBaseRDD.count())    //## 获取键值对信息    val keyValue = hBaseRDD.map(x => x._2).map(_.list)     //outPut is a RDD[String], in which each line represents a record in HBase     val outPut = keyValue.flatMap(x =>  x.asScala.map(cell =>         "columnFamily=%s,qualifier=%s,timestamp=%s,type=%s,value=%s".format(           Bytes.toStringBinary(CellUtil.cloneFamily(cell)),           Bytes.toStringBinary(CellUtil.cloneQualifier(cell)),           cell.getTimestamp.toString,           Type.codeToType(cell.getTypeByte),           Bytes.toStringBinary(CellUtil.cloneValue(cell))         )       )     )     outPut.foreach(println)

这样即可得到下面的结果。