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)
这样即可得到下面的结果。
