Spark通过Scan的API读取HBase表限定范围的数据

喜欢ヅ旅行 2023-05-31 03:25 6阅读 0赞

在这篇博客中[Spark对HBase进行数据的读写操作][Spark_HBase]，我通过代码说明如何通过Spark对HBase表的数据进行读取并转化为RDD。  
但是，这种方式只能是进行全表读取，那如果我们只想读取一部分的数据，应该如何实现呢？  
**通过hbase shell中的scan方法的API，设置起始行和结束行，对限定范围的数据进行扫描转化为RDD**

import java.text.SimpleDateFormat;
    
    import org.apache.hadoop.conf.Configuration;
    import org.apache.hadoop.hbase.HBaseConfiguration;
    import org.apache.hadoop.hbase.client.Result;
    import org.apache.hadoop.hbase.client.Scan;
    import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
    import org.apache.hadoop.hbase.mapreduce.TableInputFormat;
    import org.apache.hadoop.hbase.protobuf.ProtobufUtil;
    import org.apache.hadoop.hbase.protobuf.generated.ClientProtos;
    import org.apache.spark.SparkConf;
    import org.apache.spark.api.java.JavaSparkContext;
    
    
    public void test(String startDate, String endDate){ 
    	SparkConf conf = new SparkConf().setAppName("Test").setMaster("yarn").set("deploy-mode", "cluster");
    	JavaSparkContext sc = new JavaSparkContext(conf);
    	
    	// 设置HBase集群的配置
    	Configuration hbaseConf = HBaseConfiguration.create();
    	hbaseConf.set("hbase.zookeeper.property.clientPort", "2181");
    	hbaseConf.set("hbase.zookeeper.quorum", "master,slave1,slave2");
    	
    	SimpleDateFormat format = new SimpleDateFormat("yyyyMMdd-HH:mm:ss");
    	Scan scan = new Scan();
    	// rowkey格式为：时间戳+xxxx
    	// 在这里设置你想对HBase表扫描的起始和结束行
    	scan.setStartRow(String.valueOf(format.parse(startDate).getTime()).getBytes());
    	scan.setStopRow(String.valueOf(format.parse(endDate).getTime()+1).getBytes());
    	
    	hbaseConf.set(TableInputFormat.INPUT_TABLE, "hbase_table");
    	ClientProtos.Scan proto = ProtobufUtil.toScan(scan);
    	String ScanToString = Base64.encodeBytes(proto.toByteArray());
    	hbaseConf.set(TableInputFormat.SCAN, ScanToString);
    	
    	JavaPairRDD<ImmutableBytesWritable, Result> dwRDD = sc.newAPIHadoopRDD(hbaseConf, TableInputFormat.class,
    	           org.apache.hadoop.hbase.io.ImmutableBytesWritable.class,
    	           Result.class);
    	
    	JavaRDD<String[]> data = dwRDD.map(new Function<Tuple2<ImmutableBytesWritable,Result>, String[]>() { 
    
    			private static final long serialVersionUID = 1L;
    
    			@Override
    			public String[] call(Tuple2<ImmutableBytesWritable, Result> v1) throws Exception { 
    				// TODO Auto-generated method stub
    				Result result = v1._2;
    				String[] arr = { Bytes.toString(result.getValue("items".getBytes(), "ci".getBytes())),
    						Bytes.toString(result.getValue("items".getBytes(), "begintime".getBytes())),
    						Bytes.toString(result.getValue("items".getBytes(), "imsi".getBytes())),
    						Bytes.toString(result.getValue("items".getBytes(), "latitude".getBytes())),
    						Bytes.toString(result.getValue("items".getBytes(), "longitude".getBytes()))};
    				return arr;
    			}
    		});
    		
    	System.out.println("成功从HBase读取数据");
    }

这里的`scan.setStartRow`和`scan.setStopRow`方法是传递一个字符串，设置起始行和结束行之后，HBase只会扫描出rowkey在StartRow和StopRow之间的数据，即rowkey满足这样条件的数据：StartRow<=rowkey<=StopRow。

那么这里是如何比较大小的呢？其实就是用ASCII码来进行比较的。  
举个例子，比如0对应的ASCII码为48,1的ASCII码是49，a是229，那么0<1<a。  
对于多个字符的，都是先比较第一位，第一位比较大的就大，相同则往下一位比较。例如，“1aa”<“2”，“aa”>“a”。

在java中也有这么一个方法`compareTo`，  
例如`"a".compareTo("1")>0`，而`"0".compareTo("1")<0`

这个时候，我们在设计Hbase表的rowkey时，就可以将最重要的维度数据放在rowkey的前面，例如时间戳，这样就可以实现读取限定时间范围的数据了。  
例如，rowkey一般是时间戳加上id，那么如果我们只想读取HBase中2019-10-01到2019-11-01的数据，那么这里就可以将StartRow设置为2019-10-01对应的时间戳字符串，StopRow设置为2019-11-01对应的时间戳字符串。

上面代码的maven依赖：

<dependency>
    	      <groupId>org.apache.spark</groupId>
    	      <artifactId>spark-core_2.11</artifactId>
    	      <version>2.1.0</version>
    	  </dependency>
    	  <dependency>
    	      <groupId>org.apache.hbase</groupId>
    	      <artifactId>hbase-client</artifactId>
    	      <version>1.3.1</version>
    	  </dependency>
    	  <dependency>
    	      <groupId>org.apache.hbase</groupId>
    	      <artifactId>hbase-common</artifactId>
    	      <version>1.3.1</version>
    	  </dependency>
    	  <dependency>
    	      <groupId>org.apache.hbase</groupId>
    	      <artifactId>hbase-server</artifactId>
    	      <version>1.3.1</version>
    	  </dependency>

欢迎关注同名公众号：“我就算饿死也不做程序员”。  
交个朋友，一起交流，一起学习，一起进步。![在这里插入图片描述][20200430114429333.jpg]

[Spark_HBase]: https://blog.csdn.net/sgyuanshi/article/details/96743281
[20200430114429333.jpg]: /images/20230531/bd3d1824e4664e10862d142926c29c2f.png