本篇作為【SparkSQL程式設計】系列的第二篇部落格,為大家介紹的是DataSet概念入門以及與DataFrame的互操作。
碼字不易，先贊後看，養成習慣!
3. DataSet
Dataset是具有強類型的資料集合，需要提供對應的類型資訊。
3.1 建立
1）建立一個樣例類
scala> case class Person(name: String, age: Long)
defined class Person 複制
2）建立DataSet
scala> val caseClassDS = Seq(Person("Andy", 32)).toDS()
caseClassDS: org.apache.spark.sql.Dataset[Person] = [name: string, age: bigint] 複制
3.2 RDD轉換為DataSet
SparkSQL能夠自動将包含有case類的RDD轉換成DataFrame，case類定義了table的結構，case類屬性通過反射變成了表的列名。
1）建立一個RDD
scala> val peopleRDD = sc.textFile("/input/people.txt")
peopleRDD: org.apache.spark.rdd.RDD[String] = examples/src/main/resources/people.txt MapPartitionsRDD[3] at textFile at <console>:27 複制
2）建立一個樣例類
scala> case class Person(name: String, age: Long)
defined class Person 複制
3）将RDD轉化為DataSet
scala> peopleRDD.map(line => {val para = line.split(",");Person(para(0),para(1).trim.toInt)}).toDS() 複制
3.3 DataSet 轉換為RDD
調用rdd方法即可。
1）建立一個DataSet
scala> val DS = Seq(Person("Andy", 32)).toDS()
DS: org.apache.spark.sql.Dataset[Person] = [name: string, age: bigint] 複制
2）将DataSet轉換為RDD
scala> DS.rdd
res11: org.apache.spark.rdd.RDD[Person] = MapPartitionsRDD[15] at rdd at <console>:28 複制
4.DataFrame與DataSet的互操作
1.DataFrame轉換為DataSet
1 ) 建立一個DataFrame
scala> val df = spark.read.json("/input/people.json")
df: org.apache.spark.sql.DataFrame = [age: bigint, name: string] 複制
2）建立一個樣例類
scala> case class Person(name: String, age: Long)
defined class Person 複制
3）将DateFrame轉化為DataSet
scala> df.as[Person]
res14: org.apache.spark.sql.Dataset[Person] = [age: bigint, name: string] 複制
2.DataSet轉換為DataFrame
1）建立一個樣例類
scala> case class Person(name: String, age: Long)
defined class Person 複制
2）建立DataSet
scala> val ds = Seq(Person("Andy", 32)).toDS()
ds: org.apache.spark.sql.Dataset[Person] = [name: string, age: bigint] 複制
3）将DataSet轉化為DataFrame
scala> val df = ds.toDF
df: org.apache.spark.sql.DataFrame = [name: string, age: bigint] 複制
4）展示
scala> df.show
+----+---+
|name|age|
+----+---+
|Andy| 32|
+----+---+ 複制
4.1 DataSet轉DataFrame
這個很簡單了解，因為隻是把case class封裝成Row。
（1）導入隐式轉換
import spark.implicits._ 複制
（2）轉換
val testDF = testDS.toDF 複制
4.2 DataFrame轉DataSet
（1）導入隐式轉換
import spark.implicits._ 複制
（2）建立樣例類
case class Coltest(col1:String,col2:Int)extends Serializable //定義字段名和類型 複制
（3）轉換
val testDS = testDF.as[Coltest] 複制
這種方法就是在給出每一列的類型後，使用as方法，轉成Dataset，這在資料類型是DataFrame又需要針對各個字段處理時極為友善。在使用一些特殊的操作時，一定要加上 import spark.implicits._ 不然toDF、toDS無法使用。
本次的分享就到這裡，感興趣的朋友記得點贊，關注喲~