从scala spark中的无序数据中提取列

Question

我正在学习 scala-spark，想知道如何根据列名从无序数据中提取所需的列？详情如下-

输入数据：RDD[Array[String]]

id=1,country=USA,age=20,name=abc
name=def,country=USA,id=2,age=30
name=ghi,id=3,age=40,country=USA

所需输出：

Name,id
abc,1
def,2
ghi,3

任何帮助将不胜感激。提前致谢！

Answer 1

如果你有RDD[Array[String]]，那么你可以得到想要的数据

您可以将案例类定义为

case class Data(Name: String, Id: Long)

然后将每一行解析为case class

val df = rdd.map( row => {
  //split the line and convert to map so you can extract the data
  val data = row.split(",").map(x => (x.split("=")(0),x.split("=")(1))).toMap
  Data(data("name"), data("id").toLong)
})

转换为Dataframe并显示

df.toDF().show(false)

输出：

+----+---+
|Name|Id |
+----+---+
|abc |1  |
|def |2  |
|ghi |3  |
+----+---+

这是读取文件的完整代码

case class Data(Name: String, Id: Long)

def main(args: Array[String]): Unit = {
  val spark = SparkSession.builder().appName("xyz").master("local[*]").getOrCreate()

  import spark.implicits._
  val rdd = spark.sparkContext.textFile("path to file ")

  val df = rdd.map(row => {
    val data = row.split(",").map(x => (x.split("=")(0), x.split("=")(1))).toMap
    Data(data("name"), data("id").toLong)
  })

  df.toDF().show(false)
}