如何通过Bigtable Spark Connector读取数据时获取Cell Timestamp？

首先明确：当前Bigtable Spark Connector的catalog JSON配置方式无法直接获取Cell的时间戳，这种映射方式仅将列的最新值映射为DataFrame字段，不暴露时间戳这类元数据。

如果想要在分布式读取场景下获取Cell时间戳，可以采用以下两种方案：

方案1：基于HBase InputFormat的分布式读取（兼容Bigtable） #

Bigtable兼容HBase API，而Spark可以通过newAPIHadoopRDD实现分布式读取，这种方式天然支持任务并行，完全满足你对分布式读取的需求。具体代码示例如下：

import org.apache.hadoop.hbase.{HBaseConfiguration, TableName}
import org.apache.hadoop.hbase.client.Result
import org.apache.hadoop.hbase.io.ImmutableBytesWritable
import org.apache.hadoop.hbase.mapreduce.TableInputFormat
import org.apache.spark.sql.SparkSession

object BigtableTimestampReader {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder().appName("BigtableTimestampReader").getOrCreate()
    val sc = spark.sparkContext

    // 配置Bigtable连接参数
    val conf = HBaseConfiguration.create()
    conf.set("hbase.zookeeper.quorum", "bigtable.googleapis.com")
    conf.set("hbase.zookeeper.property.clientPort", "2181")
    conf.set("hbase.client.connection.impl", "com.google.cloud.bigtable.hbase1_x.BigtableConnection")
    conf.set(TableInputFormat.INPUT_TABLE, s"${BT_PROJECT}:${BT_INSTANCE}.table_name")

    // 分布式读取数据，返回RDD[(ImmutableBytesWritable, Result)]
    val hbaseRDD = sc.newAPIHadoopRDD(conf, classOf[TableInputFormat], classOf[ImmutableBytesWritable], classOf[Result])

    // 解析Result，提取rowkey、列值和时间戳
    val resultRDD = hbaseRDD.map { case (_, result) =>
      val rowkey = new String(result.getRow)
      // 获取指定列族和列的所有Cell（包含时间戳）
      val cells = result.getColumnCells("column_family".getBytes, "column_qualifier".getBytes)
      cells.map(cell => {
        val value = new String(cell.getValueArray, cell.getValueOffset, cell.getValueLength)
        val timestamp = cell.getTimestamp
        (rowkey, value, timestamp)
      })
    }.flatMap(identity)

    // 转换为DataFrame方便后续处理
    import spark.implicits._
    val df = resultRDD.toDF("rowKey", "value", "timestamp")
    df.show()

    spark.stop()
  }
}

方案2：扩展Bigtable Spark Connector（复杂但贴合原有使用习惯） #

如果你坚持想要用DataFrame的format("bigtable")方式读取，需要自定义Connector的读取逻辑，重写列映射逻辑以包含时间戳。但这种方式需要修改Connector的源码或者自定义UDF结合底层API，实现成本较高，不如方案1直接高效。

另外补充：你之前担心的Scala版HBase API客户端不支持分布式读取是误解，通过Spark的newAPIHadoopRDD调用HBase InputFormat时，Spark会自动将读取任务拆分到多个Executor节点执行，完全是分布式的。

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