java - 如何使用Scala(Spark)API从HBase读取列名称和值？

Question

我在hbase中有一个巨大的表，可能有数百万行。
[HBase表结构] [1]
我正在尝试使用(STARTROW＆ENDROW)和sc.newAPIHadoopRDD函数访问表的大块。
我试图找到一种方法，可以从生成的RDD中获取列限定符名称。由于每一行可以具有任意数量的列和列限定符，因此我想通过Rowkey获取每一行的列族。
简而言之，我想在Spark中创建一个看起来像这样的Dataframe:

ROWKEY  COLUMN NAME                     VALUE
ROW1    ColumnFamily:ColumnQualifier1   Value="XX"
ROW1    ColumnFamily:ColumnQualifier2   Value="XX"
ROW1    ColumnFamily:ColumnQualifier3   Value="XX"
ROW1    ColumnFamily:ColumnQualifier4   Value="XX"
ROW1    ColumnFamily:ColumnQualifier5   Value="XX"
ROW1    ColumnFamily:ColumnQualifier6   Value="XX"
ROW2    ColumnFamily:ColumnQualifier1   Value="XX"
ROW2    ColumnFamily:ColumnQualifier2   Value="XX"
ROW2    ColumnFamily:ColumnQualifier3   Value="XX"
ROW2    ColumnFamily:ColumnQualifier4   Value="XX"
ROW3    ColumnFamily:ColumnQualifier1   Value="XX"
ROW4    ColumnFamily:ColumnQualifier2   Value="XX"

因此，从 sc.newAPIHadoopRDD返回的RDD中，我想知道一种访问列名的方法。
一旦有了列限定符，就可以使用 rdd.getValue(family,qualifier)函数获得family:限定符组合的值(value)。

val kvRDD = sc.newAPIHadoopRDD(hbaseConf,classOf[TableInputFormat],classOf[ImmutableBytesWritable], classOf[Result])
val resultRDD = kvRDD.map(tuple => tuple._2)
val keyValueRDD = resultRDD.map(result => {
    var resultStrings: List[Object] = List()
    var navigablemap=result.getNoVersionMap()
    val vallist = navigablemap.values() 
    for (each <- vallist) {
        resultStrings = resultStrings ::: List(each)
    }
    resultStrings
})

但这将返回一个rdd，其中每一行都被加密。
非常感谢scala代码的帮助，
谢谢

Answer 1

val dataFrame = kvRDD.map(x => {
      val rowkey=Bytes.toString(x._2.getRow)
      val families=x._2.getNoVersionMap.keySet().iterator()
      var colums = scala.collection.mutable.Map[String, String]()
      while (families.hasNext){
        val family=families.next();
        val qualifiers=x._2.getFamilyMap(family).keySet().iterator()
        while(qualifiers.hasNext){
          val qualifier=qualifiers.next();
          colums+=((Bytes.toString(family)+":"+(Bytes.toString(qualifier))->Bytes.toString(x._2.getValue(family,qualifier))))
        }
      }
      (rowkey,colums)
    }).toDF()
    dataFrame.select($"_1".alias("ROWKEY"), functions.explode($"_2").as(List("COLUMN NAME","VALUE"))).show