Spark 2基于Scala全量扫描HBase及hbase-spark模块去向咨询

我来帮你理清这两个核心问题，结合HBase的版本演进和Spark 2的实践经验给你具体方案：

一、hbase-spark模块的最新位置 #

从HBase 2.0版本开始，hbase-spark模块已经从HBase主仓库中拆分出来，迁移到了独立的HBase Connectors项目中。这个项目专门负责维护HBase与各种计算引擎（包括Spark）的连接器实现，是官方现在主推的连接器维护载体。

你提到HBase主仓库的master分支仍有该模块，这是因为主分支保留了旧代码的快照，但正式发布的HBase 2.0+版本中，hbase-spark的维护、更新和发布都转移到Connectors项目了。如果要在Spark 2环境中使用，你需要引入对应版本的hbase-spark依赖，比如Maven配置可以参考（需匹配你的HBase和Spark版本）：

<dependency>
  <groupId>org.apache.hbase</groupId>
  <artifactId>hbase-spark</artifactId>
  <version>2.4.11</version> <!-- 对应HBase 2.4.x系列，适配Spark 2.4.x -->
</dependency>

二、Spark 2下无固定Catalog的HBase全量扫描实现 #

既然没法用SHC（缺少固定Catalog），这里给你两种可行的实现方式，都是基于Scala的实践方案：

方案1：原生HBase API + Spark RDD #

这种方式不需要额外引入连接器，直接用Spark的newAPIHadoopRDD对接HBase的MapReduce输入格式，灵活性很高：

import org.apache.hadoop.hbase.{HBaseConfiguration, TableName}
import org.apache.hadoop.hbase.client.{ConnectionFactory, Scan}
import org.apache.hadoop.hbase.io.ImmutableBytesWritable
import org.apache.hadoop.hbase.mapreduce.TableInputFormat
import org.apache.hadoop.hbase.util.Bytes
import org.apache.spark.SparkContext

object HBaseFullScan {
  def main(args: Array[String]): Unit = {
    // 初始化SparkContext（集群环境请替换为对应配置）
    val sc = new SparkContext("local[*]", "HBaseFullScanDemo")
    val hbaseConf = HBaseConfiguration.create()
    
    // 配置要扫描的HBase表名
    val targetTable = "your_hbase_table"
    hbaseConf.set(TableInputFormat.INPUT_TABLE, targetTable)
    
    // 构建Scan对象，可自定义扫描规则（比如过滤、指定列族/列）
    val scan = new Scan()
    // 示例：如果只需要读取特定列族，可添加 scan.addFamily(Bytes.toBytes("cf1"))
    // 将Scan序列化到配置中
    hbaseConf.set(TableInputFormat.SCAN, TableInputFormat.convertScanToString(scan))
    
    // 读取HBase数据为RDD
    val hbaseRDD = sc.newAPIHadoopRDD(
      hbaseConf,
      classOf[TableInputFormat],
      classOf[ImmutableBytesWritable],
      classOf[org.apache.hadoop.hbase.client.Result]
    )
    
    // 解析Result对象，处理业务逻辑
    val processedData = hbaseRDD.map { case (_, result) =>
      val rowKey = Bytes.toString(result.getRow)
      // 示例：读取cf1下col1列的值
      val columnValue = Bytes.toString(result.getValue(Bytes.toBytes("cf1"), Bytes.toBytes("col1")))
      (rowKey, columnValue)
    }
    
    // 输出或进一步处理数据
    processedData.foreach(println)
    sc.stop()
  }
}

方案2：使用hbase-spark模块的简化API #

如果你引入了前面提到的hbase-spark依赖，可以用HBaseContext来简化代码，不需要手动配置TableInputFormat：

import org.apache.hadoop.hbase.HBaseConfiguration
import org.apache.hadoop.hbase.client.Scan
import org.apache.hadoop.hbase.util.Bytes
import org.apache.spark.SparkContext
import org.apache.hadoop.hbase.spark.HBaseContext

object HBaseFullScanWithConnector {
  def main(args: Array[String]): Unit = {
    val sc = new SparkContext("local[*]", "HBaseFullScanWithConnector")
    val hbaseConf = HBaseConfiguration.create()
    val hbaseContext = new HBaseContext(sc, hbaseConf)
    
    val targetTable = "your_hbase_table"
    val scan = new Scan()
    // 优化扫描性能：设置缓存大小和批量读取数
    scan.setCaching(1000)
    scan.setBatch(100)
    
    // 直接获取HBase RDD
    val hbaseRDD = hbaseContext.hbaseRDD(TableName.valueOf(targetTable), scan)
    
    // 解析数据
    val resultRDD = hbaseRDD.map { result =>
      val rowKey = Bytes.toString(result.getRow)
      val value = Bytes.toString(result.getValue(Bytes.toBytes("cf1"), Bytes.toBytes("col1")))
      (rowKey, value)
    }
    
    resultRDD.foreach(println)
    sc.stop()
  }
}

注意事项 #

• 版本兼容性：Spark 2.4.x建议搭配HBase 2.4.x系列的hbase-spark依赖，避免版本冲突
• 配置文件：运行时要确保hbase-site.xml在Spark的classpath中，或者通过代码设置ZooKeeper地址等关键参数
• 性能优化：全量扫描时可以通过scan.setCaching()、scan.setBatch()调整批量读取大小，提升扫描效率

内容的提问来源于stack exchange，提问作者angelcervera