SparkSql------RDD、DataFrame、DataSet之间的相互转化

在老的版本中，SparkSQL提供两种SQL查询起始点，一个叫SQLContext，用于Spark自己提供的SQL查询，一个叫HiveContext，用于连接Hive的查询，SparkSession是Spark最新的SQL查询起始点，实质上是SQLContext和HiveContext的组合，所以在SQLContext和HiveContext上可用的API在SparkSession上同样是可以使用的。SparkSession内部封装了sparkContext，所以计算实际上是由sparkContext完成的。

SparkSession的三种创建方式

添加依赖

<dependency>
    <groupId>org.apache.spark</groupId>
    <artifactId>spark-sql_2.11</artifactId>
    <version>${spark.version}</version>
</dependency>
           

/**
      * 创建SparkSession
      * builder用于创建一个SparkSession。
      * appName设置App的名字
      * master设置运行模式(集群模式不用设置)
      * enableHiveSupport 开启hive操作
      * getOrCreate 进行创建
      */
val sparks1=SparkSession.builder().appName("SparkSQLDemo").master("local").getOrCreate()
//-----------------------------------------------------------------------
val conf = new SparkConf().setAppName("SparkSQLDemo").setMaster("local")
val sparks2 = SparkSession.builder().config(conf).getOrCreate()
//-----------------------------------------------------------------------
val sparkh = SparkSession.builder().appName("SparkSQLDemo").master("local")    .enableHiveSupport().getOrCreate()
 
           

RDD转换为DataFrame

1.直接手动确定

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.{DataFrame, SparkSession}

import org.apache.spark.rdd.RDD
/**
  * RDD--->DataFrame 直接手动确定
  */
object SparkSQLDemo {
  def main(args: Array[String]): Unit = {
    //创建SparkConf()并设置App名称
    val conf = new SparkConf().setAppName("SparkSQLDemo").setMaster("local")
    //SQLContext要依赖SparkContext
    val sc = new SparkContext(conf)
    //从指定的地址创建RDD
    val lineRDD = sc.textFile("person.txt").map(_.split(","))
    //这里是将数据转换为元组,数据量少可以使用这种方式
    val tuple: RDD[(String, Int)] = lineRDD.map(x => (x(0),x(1).trim().toInt))
    val spark = SparkSession.builder().config(conf).getOrCreate()
    //如果需要RDD于DataFrame之间操作,那么需要引用 import spark.implicits._ [spark不是包名,是SparkSession对象]
    import spark.implicits._
    val frame: DataFrame = tuple.toDF("name","age")
    frame.show()
    sc.stop();
  }
}
           

2.通过反射获取Scheam

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.{DataFrame, SparkSession}
/**
  * RDD--->DataFrame 通过反射推断Schema
  */
object SparkSQLDemo {
  def main(args: Array[String]): Unit = {
    //创建SparkConf()并设置App名称
    val conf = new SparkConf().setAppName("SparkSQLDemo").setMaster("local")
    //SQLContext要依赖SparkContext
    val sc = new SparkContext(conf)
    //从指定的地址创建RDD
    val lineRDD = sc.textFile("people.txt").map(_.split(","))
    //除了这个方式之外我们还可以是用样例类的形式来做复杂的转换操作
    val tuple = lineRDD.map(x => People(x(0),x(1).trim().toInt));
    val spark = SparkSession.builder().config(conf).getOrCreate()
    import spark.implicits._
    //此时不需要使用指定列名会根据样例类中定义的属性来确定类名
    val frame: DataFrame = tuple.toDF
    frame.show()
    sc.stop()
  }
}
case class People(name:String,age:Int)
           

3.通过StructType直接指定Schema

import org.apache.spark.sql.types.{IntegerType, StringType, StructField, StructType}
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.{DataFrame, Row, SparkSession}
/**
  * RDD--->DataFrame 通过StructType直接指定Schema
  */
object SparkSQLDemo {
  def main(args: Array[String]): Unit = {
    //创建SparkConf()并设置App名称
    val conf = new SparkConf().setAppName("SparkSQLDemo").setMaster("local")
    //创建SparkContext对象
    val sc = new SparkContext(conf)
    //创建SparkSession对象
    val spark = SparkSession.builder().config(conf).getOrCreate()
    //从指定的地址创建RDD
    val lineRDD = sc.textFile("people.txt").map(_.split(","))
    //通过StructType直接指定每个字段的schema
    val schema = StructType(
      List(
        StructField("name", StringType, true),
        StructField("age", IntegerType, true)
      )
    )
    //将RDD映射到rowRDD
    val rowRDD = lineRDD.map(p => Row(p(0), p(1).trim.toInt))
    //将schema信息应用到rowRDD上
    val peopleDataFrame = spark.createDataFrame(rowRDD, schema)
    val frame: DataFrame = peopleDataFrame.toDF
    frame.show()
    sc.stop()
  }
}
           

RDD转换为DataSet

通过反射获取Scheam(样例类模式)

package spark.SparkSql

import org.apache.spark.rdd.RDD
import org.apache.spark.sql.types.{IntegerType, StringType, StructField, StructType}
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.{DataFrame, Dataset, Row, SparkSession}
/**
  * RDD--->DataSet 通过反射获取Scheam
  */
object Rdd2DataSet {
  def main(args: Array[String]): Unit = {
    //创建SparkConf()并设置App名称
    val conf = new SparkConf().setAppName("SparkSQLDemo").setMaster("local")
    //创建SparkContext对象
    val sc = new SparkContext(conf)
    //创建SparkSession对象
    val spark = SparkSession.builder().config(conf).getOrCreate()
    val value: RDD[String] = sc.textFile("person.txt")
    import spark.implicits._
    val dataSet: Dataset[People] = value.map {
      x =>
        val para = x.split(",")
        People(para(0).trim().toInt, para(1))
    }.toDS()

    dataSet.show()

//    1:aaa:People1(1,aaa)
//    3:bbb:People1(3,bbb)
//    2:ccc:People1(2,ccc)

      sc.stop()
  }
}
case class People(id:Int,name:String)
           

DataSet转化到RDD

//调用rdd方法
val dataSetTORDD:RDD[People] = dataSet.rdd
dataSetTORDD.foreach(x => println(x.name+","+x.age));
           

DataSet转换DataFrame

//调用toDF方法,直接服用case class中定义的属性
val frame: DataFrame = dataSet.toDF()
frame.show()
           

DataFrame转换DataSet

val value: Dataset[People] = frame.as[People] 
case class People(name:String,age:Int)
           

总结:

SparkSQL支持两种类型分别为DataSet和DataFrame,这两种类型都支持从RDD转换为DataSet或DataFrame

RDD转DataFrame有三种方法是

    1.直接转换即使用元组的模式存储在转换

    2.使用样例类的模式匹配Scheam在转换

    3.StructType直接指定Schema在转换

ps:其余读取文件的方式可以直接获取对应的DataFrame

比如：

val conf = new SparkConf().setAppName("SparkSQLDemo").setMaster("local")
val spark = SparkSession.builder().config(conf).getOrCreate()
val df: DataFrame = spark.read.json("people.json")
           

RDD转DataSet

使用样例类的模式匹配Scheam再转换

DataSet转换DataFrame

调用toDF方法,直接用case class中定义的属性

DataFrame转换DataSet

调用as[对应样例类类名]

DataSet和DataFrame转换为RDD

DataSet对象或DataFrame对象调用rdd方法就可以转换为rdd