SnapshotScanMR速率比TableScanMR快10~30倍，那Spark如何实现SnapshotScanM
HBase 提供的 TableScanMR 和 SnapshotScanMR 是两种用于在大数据集中举行扫描的 MapReduce 作业，网上也有很多介绍Spark如何实现TableScanMR，但是对SnapshotScanMR的实现方式很少几乎没找到可用的，接下来我们先说说这两者的一些共同点以及不同的实现原理，再介绍Spark是如何实现的。雷同点
[*] 目的：
两者都是为了高效地扫描 HBase 中大量的数据，适用于数据分析和批处理使命。
[*] 基于 MapReduce：
两个作业都基于 Hadoop 的 MapReduce 框架，实现分布式盘算，能够利用集群资源举行并行处理。
[*] 支持高并发：
这两种扫描方式都能充实利用 HBase 的分布式特性，支持高并发的读操作，适合处理大规模数据集。
不同点
[*] 数据来源：
[*]TableScanMR：
直接对 HBase 实时表举行访问，读取当前的数据状态，包含所有的行和最新的变革。
[*]SnapshotScanMR：
读取特定快照的数据，快照创建时数据的一个静态视图，不会反映后续的更改。
[*] 一致性：
[*]TableScanMR：
由于是实时访问，扫描期间大概受到其他写操作的影响，因此返回的数据大概存在一致性问题。
[*]SnapshotScanMR：
提供了对某一时刻的数据访问，因此在扫描时数据是一致的，不受后续操作影响。
[*] 性能特性：
[*]TableScanMR：
性能大概受到数据库实时写入和更新的影响，适用于实时数据分析，但在写入繁重的环境中性能大概波动。
[*]SnapshotScanMR：
通常具有较稳定的性能，适合于对历史数据的分析和处理，性能受后续写入影响较小。
[*] 实现的原理：
[*] TableScanMR：
TableScan现实上还是一种并行的ScanApi，它离不开RegionServer，所有的Scan请求都会打到RegionServer上，所以假如RegionServer有压力时这种Scan效果并不理想。比如我们在Scan的同时，服务还在大量的compact 大概另有其他的bulkload的等操作影响RegionServer压力的时，Scan效果不是很理想。工作原理如下
https://i-blog.csdnimg.cn/direct/ff97cde46e5c414c908029b4cc0ca4ca.png
[*] SnapshotScanMR：
它的Scan方式是直接绕过了RegionServer，直接读取Hbase的HDFS文件，所以RegionServer的压力对他无影响。影响它的就是磁盘IO大概网络。所以当执利用命和Hbase数据是在同机房时SnapshotScan的速率是TableScan 的10~30倍，假如跨机房数据量大了大概还不如TableScan，所以得包管同机房。工作原理如下
https://i-blog.csdnimg.cn/direct/70d242cb3bf848efbecf600d8792ab78.png
[*] 使用场景：
[*]TableScanMR：
[*]适用于必要访问最新数据的场景，如实时盘算、在线分析等。
[*]适用于数据分布均匀的场景
[*]SnapshotScanMR：
[*]适用于Bulkload后直接Scan的场景
[*]适用于缓解RegionServer压力大大概压力分布不均匀的场景（前提可以忽略实时的写入数据，究竟使用的是快照）
[*]适用于大部分表数据分布均匀，但是部分大表数据分布不均匀，经常大量长时的compact影响单RegionServer压力，导致在Scan均匀表时出现部分长尾使命的情况（这种也是我所遇到的）。
[*]盘算使命和数据在同机房的情况这点很重要，假如跨机房数据复制的RPC就会非常的耗时
[*] Spark实现方式：
[*]TableScanMR：
val sc = new SparkContext(sparkConf)
val scan = new Scan();
scan.addFamily(Bytes.toBytes("c"))
scan.setTimeStamp(timeStamp)
scan.withStartRow(Bytes.toBytes(startRow))
scan.withStopRow(Bytes.toBytes(stopRow))
val hbaseConf = new Configuration()
hbaseConf.set(TableInputFormat.SCAN, Base64.encodeBytes(ProtobufUtil.toScan(scan).toByteArray))
hbaseConf.set(TableInputFormat.INPUT_TABLE, "Hbase的表名")
val dataRDD: RDD[(ImmutableBytesWritable, Result)] = sc.newAPIHadoopRDD(hbaseConf, classOf, classOf, classOf)
//处理后续逻辑
[*]SnapshotScanMR：
//1.定义扫描范围
val sc = new SparkContext(sparkConf)
val scan = new Scan();
scan.addFamily(Bytes.toBytes("c"))
scan.setTimeStamp(timeStamp)
scan.withStartRow(Bytes.toBytes(startRow))
scan.withStopRow(Bytes.toBytes(stopRow))
//2.创建快照
val table = TableName.valueOf("Hbase的表名")
val snapshotName = "快照名称"
val tmpRestoreDir = new Path("临时目录")
//创建快照（可以参考Hbase创建快照）
Admin.snapshot(snapshotName, table);
//3.创建SnapshotScan并返回一个RDD
val job: Job = Job.getInstance()
TableMapReduceUtil.initCredentials(job)
TableMapReduceUtil.initTableSnapshotMapperJob(snapshotName, scan,classOf, null, null, job, true, tmpRestoreDir)
//最后一个参数是HBaseContext 可以根据具体的实现传入
val dataRDD: RDD[(ImmutableBytesWritable, Result)] = new NewHBaseRDD(sc,classOf,classOf,classOf, job.getConfiguration,HBaseContext)
//处理后续逻辑
//4.删除快照和目录
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。
页:
[1]