Spark之RDD、DataSet、DataFrame转换

偏执的太偏执、 2022-11-28 00:52 377阅读 0赞

前面已经给大家说过RDD、DataSet、DataFrame三者之间的关系以及区别，各自都有各自的优劣。RDD、DataFrame、Dataset三者有许多共性，有各自适用的场景常常需要在三者之间转换。今天就给大家说说他们直接按的无缝转换。

## DataFrame/Dataset转RDD ##

这个转换很简单，只需要在类型的后面.rdd就可以转换成RDD。  
如何创建DataFrame/Dataset的方法在前面已经说过在这里就不多说了。

val createDS =  spark.createDataset(1 to 3)
    val createDF = spark.createDataFrame(List( ("Scala", 35),("Python", 30), ("R", 15), ("Java", 20)) )
    
    val rdd1=createDS.rdd
    val rdd2=createDF.rdd

## RDD转Dataset ##

这个属于隐式转换，需要导包才能转换，不然会报错，然后在类型的后面.toDS就可以转换成Dataset

可以注意到，定义每一行的类型（case class）时，已经给出了字段名和类型，后面只要往case class里面添加值即可

## RDD转DataFrame ##

和转DataSet类型也是需要导包才能转换

import spark.implicits._
    //方式一：通过反射获取RDD内的Schema
    case class Person(name:String,age:Int)
    val people=sc.textFile("file:///home/hadooop/data/people.txt")
      .map(_.split(","))
      .map(p => Person(p(0), p(1).trim.toInt)).toDF()
    
    //方式二：通过编程接口指定Schema
    case class Person(name:String,age:Int)
    val people=sc.textFile("file:///data/people.txt")
    // 以字符串的方式定义DataFrame的Schema信息
    val schemaString = "name age"
    //导入所需要的类
    import org.apache.spark.sql.Row
    import org.apache.spark.sql.types.{ StructType, StructField, StringType}
    // 根据自定义的字符串schema信息产生DataFrame的Schema
    val schema = StructType(schemaString.split(" ").map(fieldName =>StructField(fieldName,StringType, true)))
    //将RDD转换成Row
    val rowRDD = people.map(_.split(",")).map(p => Row(p(0), p(1).trim))
    // 将Schema作用到RDD上
    val peopleDataFrame = spark.createDataFrame(rowRDD, schema)
    // 将DataFrame注册成临时表
    peopleDataFrame.createOrReplaceTempView("people")
    val results = spark.sql("SELECT name FROM people")
    results.show

一般用元组把一行的数据写在一起，然后在toDF中指定字段名

## DataSet转DataFrame ##

这个也很简单，因为只是把case class封装成Row  
和RDD转DataFrame方法类似，使用toDF

import spark.implicits._
    val createDS =  spark.createDataset(1 to 3)
    val createDF = createDS.toDF

## DataFrame转DataSet ##

这种方法就是在给出每一列的类型后，使用as方法，转成Dataset,类型一般是样例类的类型

import spark.implicits._
    case class Point(label:String,x:Double,y:Double)
    case class Category(id:Long,name:String)
    val pointsRDD=sc.parallelize(List(("bar",3.0,5.6),("foo",-1.0,3.0)))
    val categoriesRDD=sc.parallelize(List((1,"foo"),(2,"bar")))
    val points=pointsRDD.map(line=>Point(line._1,line._2,line._3)).toDS
    val categories=categories.map(line=>Category(line._1,line._2)).toDS
    val points=pointsRDD.map(line=>Point(line._1,line._2,line._3)).toDF
    val categories=categoriesRDD.map(line=>Category(line._1,line._2)).toDF
    val pointDF = points.as[Point]
    val categoryDF = categories.as[Category]

**注意：**  
在使用隐式转换的操作时，一定要加上 import spark.implicits.\_ 不然toDF、toDS无法使用