SnapshotScanMR速度比TableScanMR快10~30倍，那Spark如何实现SnapshotScanMR

HBase 提供的 TableScanMR 和 SnapshotScanMR 是两种用于在大数据集中进行扫描的 MapReduce 作业，网上也有很多介绍Spark如何实现TableScanMR，但是对SnapshotScanMR的实现方式很少几乎没找到可用的，接下来我们先说说这两者的一些共同点以及不同的实现原理，再介绍Spark是如何实现的。

相同点

1. 目的：
   两者都是为了高效地扫描 HBase 中大量的数据，适用于数据分析和批处理任务。

2. 基于 MapReduce：
   两个作业都基于 Hadoop 的 MapReduce 框架，实现分布式计算，能够利用集群资源进行并行处理。

3. 支持高并发：
   这两种扫描方式都能充分利用 HBase 的分布式特性，支持高并发的读操作，适合处理大规模数据集。

不同点

1. 数据来源：

   • TableScanMR：
     直接对 HBase 实时表进行访问，读取当前的数据状态，包含所有的行和最新的变化。
   • SnapshotScanMR：
     读取特定快照的数据，快照创建时数据的一个静态视图，不会反映后续的更改。

2. 一致性：

   • TableScanMR：
     由于是实时访问，扫描期间可能受到其他写操作的影响，因此返回的数据可能存在一致性问题。
   • SnapshotScanMR：
     提供了对某一时刻的数据访问，因此在扫描时数据是一致的，不受后续操作影响。

3. 性能特性：

   • TableScanMR：
     性能可能受到数据库实时写入和更新的影响，适用于实时数据分析，但在写入繁重的环境中性能可能波动。
   • SnapshotScanMR：
     通常具有较稳定的性能，适合于对历史数据的分析和处理，性能受后续写入影响较小。

4. 实现的原理：

   • TableScanMR：
     TableScan实际上还是一种并行的ScanApi，它离不开RegionServer，所有的Scan请求都会打到RegionServer上，所以如果RegionServer有压力时这种Scan效果并不理想。比如我们在Scan的同时，服务还在大量的compact 或者还有其他的bulkload的等操作影响RegionServer压力的时，Scan效果不是很理想。工作原理如下
     

   • SnapshotScanMR：
     它的Scan方式是直接绕过了RegionServer，直接读取Hbase的HDFS文件，所以RegionServer的压力对他无影响。影响它的就是磁盘IO或者网络。所以当执行任务和Hbase数据是在同机房时SnapshotScan的速度是TableScan 的10~30倍，如果跨机房数据量大了可能还不如TableScan，所以得保证同机房。工作原理如下
     

5. 使用场景：

   • TableScanMR：
     • 适用于需要访问最新数据的场景，如实时计算、在线分析等。
     • 适用于数据分布均匀的场景
   • SnapshotScanMR：
     • 适用于Bulkload后直接Scan的场景
     • 适用于缓解RegionServer压力大或者压力分布不均匀的场景（前提可以忽略实时的写入数据，毕竟使用的是快照）
     • 适用于大部分表数据分布均匀，但是部分大表数据分布不均匀，经常大量长时的compact影响单RegionServer压力，导致在Scan均匀表时出现部分长尾任务的情况（这种也是我所遇到的）。
     • 计算任务和数据在同机房的情况这点很重要，如果跨机房数据复制的RPC就会非常的耗时

6. 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[TableInputFormat], classOf[ImmutableBytesWritable], classOf[Result])//  处理后续逻辑

• 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并返回一个RDDval job: Job = Job.getInstance()TableMapReduceUtil.initCredentials(job)TableMapReduceUtil.initTableSnapshotMapperJob(snapshotName, scan,classOf[IdentityTableMapper], null, null, job, true, tmpRestoreDir)//最后一个参数是HBaseContext 可以根据具体的实现传入val dataRDD: RDD[(ImmutableBytesWritable, Result)] = new NewHBaseRDD(sc,classOf[TableSnapshotInputFormat],classOf[ImmutableBytesWritable],classOf[Result], job.getConfiguration,HBaseContext)//  处理后续逻辑//4.删除快照和目录