从hbase检索数据并将其格式化为scalaDataframe

cu6pst1q 于 2021-06-09 发布在 Hbase

关注(0)|答案(1)|浏览(326)

我正试图从hbase表中获取数据到apachespark环境中，但我不知道如何格式化它。有人能帮帮我吗。

case class systems( rowkey: String, iacp: Option[String], temp: Option[String])
type Record = (String, Option[String], Option[String])
val hBaseRDD_iacp = sc.hbaseTable[Record]("test_table").select("iacp","temp").inColumnFamily("test_fam") 
scala> hBaseRDD_iacp.map(x => systems(x._1,x._2,x._3)).toDF().show()
+--------------+-----------------+--------------------+
|        rowkey|             iacp|                temp|
+--------------+-----------------+--------------------+
|           ab7|0.051,0.052,0.055| 17.326,17.344,17.21|
|           k6c|   0.056,NA,0.054|17.277,17.283,17.256|
|            ad|          NA,23.0|           24.0,23.6|
+--------------+-----------------+--------------------+

但是，我希望它的格式如下。每一个逗号分隔的值都在新行中，并且每一个na都被空值替换。iacp和temp列中的值应为浮点型。每行可以有不同数量的逗号分隔值。
提前谢谢！

+--------------+-----------------+--------------------+
|        rowkey|             iacp|                temp|
+--------------+-----------------+--------------------+
|           ab7|            0.051|              17.326|
|           ab7|            0.052|              17.344|
|           ab7|            0.055|               17.21|
|           k6c|            0.056|              17.277|
|           k6c|             null|              17.283|
|           k6c|            0.054|              17.256|
|            ad|             null|                24.0|
|            ad|               23|                26.0|
+--------------+-----------------+--------------------+

hbase scala apache-spark apache-spark-sql

来源：https://stackoverflow.com/questions/55883139/retrieve-and-format-data-from-hbase-to-scala-dataframe

1条答案

按热度按时间

mwngjboj1#

你的 hBaseRDD_iacp.map(x => systems(x._1, x._2, x._3)).toDF 代码行应生成与以下内容等效的Dataframe：

val df = Seq(
  ("ab7", Some("0.051,0.052,0.055"), Some("17.326,17.344,17.21")),
  ("k6c", Some("0.056,NA,0.054"), Some("17.277,17.283,17.256")),
  ("ad", Some("NA,23.0"), Some("24.0,23.6"))
).toDF("rowkey", "iacp", "temp")

要将数据集转换为所需的结果，可以应用一个udf，将数据集的元素配对 iacp 以及 temp csv字符串以生成 (Option[Double], Option[Double]) 那是什么 explode -如下图所示：

import org.apache.spark.sql.functions._
import spark.implicits._
def pairUpCSV = udf{ (s1: String, s2: String) =>
  import scala.util.Try
  def toNumericArr(csv: String) = csv.split(",").map{
    case s if Try(s.toDouble).isSuccess => Some(s)
    case _ => None
  }
  toNumericArr(s1).zipAll(toNumericArr(s2), None, None)
}
df.
  withColumn("csv_pairs", pairUpCSV($"iacp", $"temp")).
  withColumn("csv_pair", explode($"csv_pairs")).
  select($"rowkey", $"csv_pair._1".as("iacp"), $"csv_pair._2".as("temp")).
  show(false)
// +------+-----+------+
// |rowkey|iacp |temp  |
// +------+-----+------+
// |ab7   |0.051|17.326|
// |ab7   |0.052|17.344|
// |ab7   |0.055|17.21 |
// |k6c   |0.056|17.277|
// |k6c   |null |17.283|
// |k6c   |0.054|17.256|
// |ad    |null |24.0  |
// |ad    |23.0 |23.6  |
// +------+-----+------+

注意这个值 NA 属于方法中的默认大小写 toNumericArr 因此，这并不是一个单独的案例。也， zipAll （而不是 zip )在自定义项中用于涵盖 iacp 以及 temp csv字符串具有不同的元素大小。

展开查看全部

赞(0）回复(0）举报 2021-06-09

我来回答

从hbase检索数据并将其格式化为scalaDataframe

1条答案

相关问题

热门标签

最新问答