Spark如何读取Hbase特定查询的数据

本是古典何须时尚 2022-04-06 09:11 589阅读 0赞

最近工作需要使用到Spark操作Hbase，上篇文章已经写了如何使用Spark读写Hbase全量表的数据做处理，但这次有所不同，这次的需求是Scan特定的Hbase的数据然后转换成RDD做后续处理，简单的使用Google查询了一下，发现实现方式还是比较简单的，用的还是Hbase的TableInputFormat相关的API。  
  
基础软件版本如下：

Hadoop2.7.2
    Hbase1.2.0
    Spark2.1.0
    Scala2.11.8

直接上代码如下：

`           val startRowkey="row1" 
                val endRowkey="row1"
                //开始rowkey和结束一样代表精确查询某条数据
    
                //组装scan语句
                val scan=new Scan(Bytes.toBytes(startRowkey),Bytes.toBytes(endRowkey))
                scan.setCacheBlocks(false)
                scan.addFamily(Bytes.toBytes("ks"));
                scan.addColumn(Bytes.toBytes("ks"), Bytes.toBytes("data"))
    
                //将scan类转化成string类型
                val scan_str= TableMapReduceUtil.convertScanToString(scan)
                conf.set(TableInputFormat.SCAN,scan_str)
    
                //使用new hadoop api，读取数据，并转成rdd
                val rdd = sc.newAPIHadoopRDD(conf, classOf[TableInputFormat], classOf[ImmutableBytesWritable], classOf[Result])
    
                 //打印扫描的数据总量
                 println("count:"+rdd.count)

上面的少量代码，已经完整实现了使用spark查询hbase特定的数据，然后统计出数量最后输出，当然上面只是一个简单的例子，重要的是能把hbase数据转换成RDD，只要转成RDD我们后面就能进行非常多的过滤操作。  
  
  
注意上面的hbase版本比较新，如果是比较旧的hbase，如果自定义下面的方法将scan对象给转成字符串，代码如下：

import java.io.{DataOutputStream, ByteArrayOutputStream}
    import java.lang.String
    import org.apache.hadoop.hbase.client.Scan
    import org.apache.hadoop.hbase.HBaseConfiguration
    import org.apache.hadoop.hbase.io.ImmutableBytesWritable
    import org.apache.hadoop.hbase.client.Result
    import org.apache.hadoop.hbase.mapreduce.TableInputFormat
    import org.apache.hadoop.hbase.util.Base64
    
    def convertScanToString(scan: Scan): String = {
       
      val out: ByteArrayOutputStream = new ByteArrayOutputStream
      val dos: DataOutputStream = new DataOutputStream(out)
      scan.write(dos)
      Base64.encodeBytes(out.toByteArray)
    }

最后，还有一点，上面的代码是直接自己new了一个scan对象进行组装，当然我们还可以不自己new对象，全部使用TableInputFormat下面的相关的常量，并赋值，最后执行的时候TableInputFormat会自动帮我们组装scan对象这一点通过看TableInputFormat的源码就能明白：

private static final Log LOG = LogFactory.getLog(TableInputFormat.class);
        public static final String INPUT_TABLE = "hbase.mapreduce.inputtable";
        private static final String SPLIT_TABLE = "hbase.mapreduce.splittable";
        public static final String SCAN = "hbase.mapreduce.scan";
        public static final String SCAN_ROW_START = "hbase.mapreduce.scan.row.start";
        public static final String SCAN_ROW_STOP = "hbase.mapreduce.scan.row.stop";
        public static final String SCAN_COLUMN_FAMILY = "hbase.mapreduce.scan.column.family";
        public static final String SCAN_COLUMNS = "hbase.mapreduce.scan.columns";
        public static final String SCAN_TIMESTAMP = "hbase.mapreduce.scan.timestamp";
        public static final String SCAN_TIMERANGE_START = "hbase.mapreduce.scan.timerange.start";
        public static final String SCAN_TIMERANGE_END = "hbase.mapreduce.scan.timerange.end";
        public static final String SCAN_MAXVERSIONS = "hbase.mapreduce.scan.maxversions";
        public static final String SCAN_CACHEBLOCKS = "hbase.mapreduce.scan.cacheblocks";
        public static final String SCAN_CACHEDROWS = "hbase.mapreduce.scan.cachedrows";
        public static final String SCAN_BATCHSIZE = "hbase.mapreduce.scan.batchsize";
        public static final String SHUFFLE_MAPS = "hbase.mapreduce.inputtable.shufflemaps";

上面代码中的常量，都可以conf.set的时候进行赋值，最后任务运行的时候会自动转换成scan，有兴趣的朋友可以自己尝试。  
  
\[b\]\[color=green\]\[size=large\] 有什么问题可以扫码关注微信公众号：我是攻城师(woshigcs)，在后台留言咨询。 技术债不能欠，健康债更不能欠， 求道之路，与君同行。 \[/size\]\[/color\]\[/b\] \[img\]http://dl2.iteye.com/upload/attachment/0104/9948/3214000f-5633-3c17-a3d7-83ebda9aebff.jpg\[/img\]