如何通过HBase Spark Connector获取HBase表的所有版本数据

环境信息 #

• Cloudera发行版：7.1.8
• HBase版本：2.4
• Spark版本：3.3

问题详情 #

HBase表test中，row1的cf:data列存在3个版本数据（value1、value2、value3）。执行以下Spark代码时，即便配置了hbase.spark.query.maxVersions=3，仍仅返回最新版本value3，无法获取全部3条版本数据：

val hbase_column_mapping = "device String :key, data STRING cf:data"
val hbase_table = "test"

val df = spark.read.format("org.apache.hadoop.hbase.spark")
.option("hbase.columns.mapping", hbase_column_mapping)
.option("hbase.table", hbase_table)
.option("hbase.spark.query.maxVersions", 3)
.option("hbase.spark.use.hbasecontext", false)
.load()

df.show()

当前输出：

+------+------+
|  data|device|
+------+------+
|value3|  row1|
+------+------+

期望输出：

+------+------+
|  data|device|
+------+------+
|value1|  row1|
|value2|  row1|
|value3|  row1|
+------+------+

解决方案 #

方法1：使用HBaseContext手动扫描（推荐） #

通过HBaseContext直接配置Scan获取所有版本，再将结果转换为DataFrame，这种方式更灵活可控：

import org.apache.hadoop.hbase.{HBaseConfiguration, TableName}
import org.apache.hadoop.hbase.client.Scan
import org.apache.hadoop.hbase.spark.HBaseContext
import org.apache.hadoop.hbase.util.Bytes

// 初始化HBase配置与上下文
val conf = HBaseConfiguration.create()
val hbaseContext = new HBaseContext(spark.sparkContext, conf)

// 配置Scan，设置最大版本数并指定目标列
val scan = new Scan()
scan.setMaxVersions(3)
scan.addColumn(Bytes.toBytes("cf"), Bytes.toBytes("data"))

// 从HBase读取RDD并转换为多版本DataFrame
val hbaseRDD = hbaseContext.hbaseRDD(TableName.valueOf("test"), scan)
val resultDF = hbaseRDD.flatMap { case (rowKey, result) =>
  // 提取列的所有版本数据
  val cellList = result.getColumnCells(Bytes.toBytes("cf"), Bytes.toBytes("data"))
  cellList.map { cell =>
    (Bytes.toString(rowKey), Bytes.toString(cell.getValueArray, cell.getValueOffset, cell.getValueLength))
  }
}.toDF("device", "data")

resultDF.show()

方法2：调整DataFrame Reader配置 #

若坚持使用org.apache.hadoop.hbase.spark格式读取，需启用HBaseContext并在列映射中包含时间戳字段，让连接器正确解析多版本：

val hbase_column_mapping = "device String :key, data STRING cf:data, ts Long cf:data:timestamp"
val hbase_table = "test"

val df = spark.read.format("org.apache.hadoop.hbase.spark")
  .option("hbase.columns.mapping", hbase_column_mapping)
  .option("hbase.table", hbase_table)
  .option("hbase.spark.query.maxVersions", 3)
  .option("hbase.spark.use.hbasecontext", true) // 必须启用HBaseContext
  .load()

// 若仅需value和rowkey，可按需过滤字段
df.select("device", "data").show()

原因说明 #

当hbase.spark.use.hbasecontext=false时，默认的DataFrame Reader采用批量读取优化，会忽略maxVersions参数，仅返回最新版本数据。启用HBaseContext或手动配置Scan，才能强制扫描并返回所有版本的列数据。

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