Spark SQL：RDD、DataFrames、DataSet之間的轉換

文章目錄

• RDD轉DataFrames
• 方式一：直接指定列名和數據類型
• 方式二：通過反射轉換
• 方式三：通過編程設置Schema（StructType）
• RDD轉DataSet
• DataFrame/Dataset轉RDD
• DataFrame轉Dataset
• Dataset轉DataFrame

people.txt

Michael,29
Andy,30
Justin,19

RDD轉DataFrames

scala> val rdd=sc.textFile("people.txt")
rdd: org.apache.spark.rdd.RDD[String] = people.txt MapPartitionsRDD[44] at textFile at <console>:24

方式一：直接指定列名和數據類型

scala> val ds=rdd.map(_.split(",")).map(x=>(x(0),x(1).trim().toInt)).toDF("name","age")
ds: org.apache.spark.sql.DataFrame = [name: string, age: int]

方式二：通過反射轉換

scala> case class people(name:String,age:Long)
defined class people

scala> rdd.map(_.split(",")).map(x=>(people(x(0),x(1).trim.toInt))).toDF()
res44: org.apache.spark.sql.DataFrame = [name: string, age: bigint]

方式三：通過編程設置Schema（StructType）

# 在一些時候不能直接定義case類，就用這種方法
scala> val rdd=sc.textFile("people.txt")
rdd: org.apache.spark.rdd.RDD[String] = people.txt MapPartitionsRDD[97] at textFile at <console>:27

scala> val schemaString = "name age"
schemaString: String = name age

scala> import org.apache.spark.sql.types._
import org.apache.spark.sql.types._

scala> val fields = schemaString.split(" ").map(fieldName => StructField(fieldName, StringType, nullable = true))
fields: Array[org.apache.spark.sql.types.StructField] = Array(StructField(name,StringType,true), StructField(age,StringType,true))

scala> val schems=StructType(fields)
schems: org.apache.spark.sql.types.StructType = StructType(StructField(name,StringType,true), StructField(age,StringType,true))

scala> import org.apache.spark.sql._
import org.apache.spark.sql._

scala> val rowrdd=rdd.map(_.split(",")).map(attributes => Row(attributes(0), attributes(1).trim))
rowrdd: org.apache.spark.rdd.RDD[org.apache.spark.sql.Row] = MapPartitionsRDD[99] at map at <console>:35

scala> spark.createDataFrame(rowrdd,schems)
res46: org.apache.spark.sql.DataFrame = [name: string, age: string]

RDD轉DataSet

scala> rdd.map(_.split(",")).map(x=>(x(0),x(1).trim().toInt)).toDS()
res17: org.apache.spark.sql.Dataset[(String, Int)] = [_1: string, _2: int]

DataFrame/Dataset轉RDD

scala> ds.rdd
scala> df.rdd

DataFrame轉Dataset

scala> case class people(name:String,age:Long)
defined class people

scala> df.as[people]
res39: org.apache.spark.sql.Dataset[people] = [age: bigint, name: string]

Dataset轉DataFrame

scala> ds.toDF()