从列表行键创建 Spark DataFrame

Question

我在表单或 Array[Row] 中有一个 HBase 行键列表，并希望使用这些 RowKeys 从 HBase 获取的行中创建一个 Spark DataFrame。

我在想这样的事情：

def getDataFrameFromList(spark: SparkSession, rList : Array[Row]): DataFrame = {

  val conf = HBaseConfiguration.create()
  val mlRows : List[RDD[String]] = new ArrayList[RDD[String]]

  conf.set("hbase.zookeeper.quorum", "dev.server")
  conf.set("hbase.zookeeper.property.clientPort", "2181")
  conf.set("zookeeper.znode.parent","/hbase-unsecure")
  conf.set(TableInputFormat.INPUT_TABLE, "hbase_tbl1")

  rList.foreach( r => {
    var rStr = r.toString()
    conf.set(TableInputFormat.SCAN_ROW_START, rStr)
    conf.set(TableInputFormat.SCAN_ROW_STOP, rStr + "_")
    // read one row
    val recsRdd = readHBaseRdd(spark, conf)
    mlRows.append(recsRdd)
  })

  // This works, but it is only one row
  //val resourcesDf = spark.read.json(recsRdd) 

  var resourcesDf = <Code here to convert List[RDD[String]] to DataFrame>
  //resourcesDf
  spark.emptyDataFrame
}

我可以在 for 循环中执行 recsRdd.collect() 并将其转换为字符串并将该 json 附加到 ArrayList[String 但不确定它是否有效，在这样的 for 循环中调用 collect()。

readHBaseRdd 正在使用newAPIHadoopRDD 从 HBase 获取数据

def readHBaseRdd(spark: SparkSession, conf: Configuration) = {
    val hBaseRDD = spark.sparkContext.newAPIHadoopRDD(conf, classOf[TableInputFormat],
      classOf[ImmutableBytesWritable],
      classOf[Result])

    hBaseRDD.map {
      case (_: ImmutableBytesWritable, value: Result) =>
          Bytes.toString(value.getValue(Bytes.toBytes("cf"),
                                            Bytes.toBytes("jsonCol")))
        }
    }
  }

Answer 1

使用 spark.union([mainRdd, recsRdd]) 代替列表或 RDD (mlRows)

为什么只从 HBase 读取一行？尽量有最大的间隔。

始终避免调用collect()，仅用于调试/测试。