2021年大数据Spark(二十五):SparkSQL的RDD、DF、DS相关操作

举报
Lansonli 发表于 2021/09/29 01:30:48 2021/09/29
【摘要】   目录  RDD、DF、DS相关操作 SparkSQL初体验 SparkSession 应用入口 获取DataFrame/DataSet 使用样例类 指定类型+列名 自定义Schema ​​​​​​​RDD、DF、DS相互转换 RDD、DF、DS相关操作 SparkSQL初体验 Spark...

 

目录

 RDD、DF、DS相关操作

SparkSQL初体验

SparkSession 应用入口

获取DataFrame/DataSet

使用样例类

指定类型+列名

自定义Schema

​​​​​​​RDD、DF、DS相互转换


RDD、DF、DS相关操作

SparkSQL初体验

Spark 2.0开始,SparkSQL应用程序入口为SparkSession,加载不同数据源的数据,封装到DataFrame/Dataset集合数据结构中,使得编程更加简单,程序运行更加快速高效。

 

 

SparkSession 应用入口

SparkSession:这是一个新入口,取代了原本的SQLContext与HiveContext。对于DataFrame API的用户来说,Spark常见的混乱源头来自于使用哪个“context”。现在使用SparkSession,它作为单个入口可以兼容两者,注意原本的SQLContext与HiveContext仍然保留,以支持向下兼容。

文档:http://spark.apache.org/docs/2.4.5/sql-getting-started.html#starting-point-sparksession

 1)、添加MAVEN依赖


  
  1. <dependency>
  2.     <groupId>org.apache.spark</groupId>
  3.     <artifactId>spark-sql_2.11</artifactId>
  4.     <version>2.4.5</version>
  5. </dependency>

 

 

2)、SparkSession对象实例通过建造者模式构建,代码如下:

 

 

其中①表示导入SparkSession所在的包,②表示建造者模式构建对象和设置属性,③表示导入SparkSession类中implicits对象object中隐式转换函数

 3)、范例演示:构建SparkSession实例,加载文本数据,统计条目数。


  
  1. package cn.itcast.sql
  2. import org.apache.spark.SparkContext
  3. import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}
  4. /**
  5.  * Author itcast
  6.  * Desc 演示SparkSQL
  7.  */
  8. object SparkSQLDemo00_hello {
  9.   def main(args: Array[String]): Unit = {
  10.     //1.准备SparkSQL开发环境
  11.     println(this.getClass.getSimpleName)
  12.     println(this.getClass.getSimpleName.stripSuffix("$"))
  13.     val spark: SparkSession = SparkSession.builder().appName(this.getClass.getSimpleName.stripSuffix("$")).master("local[*]").getOrCreate()
  14.     val sc: SparkContext = spark.sparkContext
  15.     sc.setLogLevel("WARN")
  16.     val df1: DataFrame = spark.read.text("data/input/text")
  17.     val df2: DataFrame = spark.read.json("data/input/json")
  18.     val df3: DataFrame = spark.read.csv("data/input/csv")
  19.     val df4: DataFrame = spark.read.parquet("data/input/parquet")
  20.     df1.printSchema()
  21.     df1.show(false)
  22.     df2.printSchema()
  23.     df2.show(false)
  24.     df3.printSchema()
  25.     df3.show(false)
  26.     df4.printSchema()
  27.     df4.show(false)
  28.     df1.coalesce(1).write.mode(SaveMode.Overwrite).text("data/output/text")
  29.     df2.coalesce(1).write.mode(SaveMode.Overwrite).json("data/output/json")
  30.     df3.coalesce(1).write.mode(SaveMode.Overwrite).csv("data/output/csv")
  31.     df4.coalesce(1).write.mode(SaveMode.Overwrite).parquet("data/output/parquet")
  32.     //关闭资源
  33.     sc.stop()
  34.     spark.stop()
  35.   }
  36. }

 

使用SparkSession加载数据源数据,将其封装到DataFrame或Dataset中,直接使用show函数就可以显示样本数据(默认显示前20条)。

Spark2.0使用全新的SparkSession接口替代Spark1.6中的SQLContext及HiveContext接口来实现其对数据加载、转换、处理等功能。SparkSession实现了SQLContext及HiveContext所有功能。 SparkSession支持从不同的数据源加载数据,并把数据转换成DataFrame,并且支持把DataFrame转换成SQLContext自身中的表,然后使用SQL语句来操作数据。SparkSession亦提供了HiveQL以及其他依赖于Hive的功能的支持。

 

获取DataFrame/DataSet

     实际项目开发中,往往需要将RDD数据集转换为DataFrame,本质上就是给RDD加上Schema信息,官方提供两种方式:类型推断和自定义Schema。

官方文档:http://spark.apache.org/docs/2.4.5/sql-getting-started.html#interoperating-with-rdds

 

 

​​​​​​​使用样例类

当RDD中数据类型CaseClass样例类时,通过反射Reflecttion获取属性名称和类型,构建Schema,应用到RDD数据集,将其转换为DataFrame。


  
  1. package cn.itcast.sql
  2. import org.apache.spark.SparkContext
  3. import org.apache.spark.rdd.RDD
  4. import org.apache.spark.sql.{DataFrame, SparkSession}
  5. /**
  6.  * Author itcast
  7.  * Desc 演示基于RDD创建DataFrame--使用样例类
  8.  */
  9. object CreateDataFrameDemo1 {
  10.   case class Person(id:Int,name:String,age:Int)
  11.   def main(args: Array[String]): Unit = {
  12.     //1.准备环境-SparkSession
  13.     val spark: SparkSession = SparkSession.builder().appName("SparkSQL").master("local[*]").getOrCreate()
  14.     val sc: SparkContext = spark.sparkContext
  15.     sc.setLogLevel("WARN")
  16.     //2.加载数据
  17.     val lines: RDD[String] = sc.textFile("data/input/person.txt")
  18.     //3.切割
  19.     //val value: RDD[String] = lines.flatMap(_.split(" "))//错误的
  20.     val linesArrayRDD: RDD[Array[String]] = lines.map(_.split(" "))
  21.     //4.将每一行(每一个Array)转为样例类(相当于添加了Schema)
  22.     val personRDD: RDD[Person] = linesArrayRDD.map(arr=>Person(arr(0).toInt,arr(1),arr(2).toInt))
  23.     //5.将RDD转为DataFrame(DF)
  24.     //注意:RDD的API中没有toDF方法,需要导入隐式转换!
  25.     import spark.implicits._
  26.     val personDF: DataFrame = personRDD.toDF
  27.     //6.查看约束
  28.     personDF.printSchema()
  29.     //7.查看分布式表中的数据集
  30.     personDF.show(6,false)//false表示不截断列名,也就是列名很长的时候不会用...代替
  31.   }
  32. }

 

此种方式要求RDD数据类型必须为CaseClass,转换的DataFrame中字段名称就是CaseClass中属性名称。

​​​​​​​指定类型+列名

除了上述两种方式将RDD转换为DataFrame以外,SparkSQL中提供一个函数:toDF,通过指定列名称,将数据类型为元组的RDD或Seq转换为DataFrame,实际开发中也常常使用。

 


  
  1. package cn.itcast.sql
  2. import org.apache.spark.SparkContext
  3. import org.apache.spark.rdd.RDD
  4. import org.apache.spark.sql.{DataFrame, SparkSession}
  5. /**
  6.  * Author itcast
  7.  * Desc 演示基于RDD创建DataFrame--使用类型加列名
  8.  */
  9. object CreateDataFrameDemo2 {
  10.   def main(args: Array[String]): Unit = {
  11.     //1.准备环境-SparkSession
  12.     val spark: SparkSession = SparkSession.builder().appName("SparkSQL").master("local[*]").getOrCreate()
  13.     val sc: SparkContext = spark.sparkContext
  14.     sc.setLogLevel("WARN")
  15.     //2.加载数据
  16.     val lines: RDD[String] = sc.textFile("data/input/person.txt")
  17.     //3.切割
  18.     //val value: RDD[String] = lines.flatMap(_.split(" "))//错误的
  19.     val linesArrayRDD: RDD[Array[String]] = lines.map(_.split(" "))
  20.     //4.将每一行(每一个Array)转为三元组(相当于有了类型!)
  21.     val personWithColumnsTypeRDD: RDD[(Int, String, Int)] = linesArrayRDD.map(arr=>(arr(0).toInt,arr(1),arr(2).toInt))
  22.     //5.将RDD转为DataFrame(DF)并指定列名
  23.     //注意:RDD的API中没有toDF方法,需要导入隐式转换!
  24.     import spark.implicits._
  25.     val personDF: DataFrame = personWithColumnsTypeRDD.toDF("id","name","age")
  26.     //6.查看约束
  27.     personDF.printSchema()
  28.     //7.查看分布式表中的数据集
  29.     personDF.show(6,false)//false表示不截断列名,也就是列名很长的时候不会用...代替
  30.   }
  31. }

 

​​​​​​​自定义Schema

依据RDD中数据自定义Schema,类型为StructType,每个字段的约束使用StructField定义,具体步骤如下:

 第一步、RDD中数据类型为Row:RDD[Row]

 第二步、针对Row中数据定义Schema:StructType

 第三步、使用SparkSession中方法将定义的Schema应用到RDD[Row]上;


  
  1. package cn.itcast.sql
  2. import org.apache.spark.SparkContext
  3. import org.apache.spark.rdd.RDD
  4. import org.apache.spark.sql.types.{IntegerType, LongType, StringType, StructField, StructType}
  5. import org.apache.spark.sql.{DataFrame, Row, SparkSession}
  6. /**
  7.  * Author itcast
  8.  * Desc 演示基于RDD创建DataFrame--使用StructType
  9.  */
  10. object CreateDataFrameDemo3 {
  11.   def main(args: Array[String]): Unit = {
  12.     //1.准备环境-SparkSession
  13.     val spark: SparkSession = SparkSession.builder().appName("SparkSQL").master("local[*]").getOrCreate()
  14.     val sc: SparkContext = spark.sparkContext
  15.     sc.setLogLevel("WARN")
  16.     //2.加载数据
  17.     val lines: RDD[String] = sc.textFile("data/input/person.txt")
  18.     //3.切割
  19.     //val value: RDD[String] = lines.flatMap(_.split(" "))//错误的
  20.     val linesArrayRDD: RDD[Array[String]] = lines.map(_.split(" "))
  21.     //4.将每一行(每一个Array)转为Row
  22.     val rowRDD: RDD[Row] = linesArrayRDD.map(arr=>Row(arr(0).toInt,arr(1),arr(2).toInt))
  23.     //5.将RDD转为DataFrame(DF)并指定列名
  24.     //注意:RDD的API中没有toDF方法,需要导入隐式转换!
  25.     import spark.implicits._
  26.     /*val schema: StructType = StructType(
  27.           StructField("id", IntegerType, false) ::
  28.           StructField("name", StringType, false) ::
  29.           StructField("age", IntegerType, false) :: Nil)*/
  30.     val schema: StructType = StructType(List(
  31.       StructField("id", IntegerType, false),
  32.       StructField("name", StringType, false),
  33.       StructField("age", IntegerType, false)
  34.     ))
  35.     val personDF: DataFrame = spark.createDataFrame(rowRDD,schema)
  36.     //6.查看约束
  37.     personDF.printSchema()
  38.     //7.查看分布式表中的数据集
  39.     personDF.show(6,false)//false表示不截断列名,也就是列名很长的时候不会用...代替
  40.   }
  41. }

此种方式可以更加体会到DataFrame = RDD[Row] + Schema组成,在实际项目开发中灵活的选择方式将RDD转换为DataFrame。

 

​​​​​​​RDD、DF、DS相互转换

实际项目开发中,常常需要对RDD、DataFrame及Dataset之间相互转换,其中要点就是Schema约束结构信息。

 1)、RDD转换DataFrame或者Dataset

转换DataFrame时,定义Schema信息,两种方式

转换为Dataset时,不仅需要Schema信息,还需要RDD数据类型为CaseClass类型

 2)、Dataset或DataFrame转换RDD

由于Dataset或DataFrame底层就是RDD,所以直接调用rdd函数即可转换

dataframe.rdd 或者dataset.rdd

 3)、DataFrame与Dataset之间转换

由于DataFrame为Dataset特例,所以Dataset直接调用toDF函数转换为DataFrame

当将DataFrame转换为Dataset时,使用函数as[Type],指定CaseClass类型即可。

 

 

 

RDD、DataFrame和DataSet之间的转换如下,假设有个样例类:case class Emp(name: String),相互转换


  
  1. RDD转换到DataFramerdd.toDF(“name”)
  2. RDD转换到Datasetrdd.map(x => Emp(x)).toDS
  3. DataFrame转换到Datasetdf.as[Emp]
  4. DataFrame转换到RDDdf.rdd
  5. Dataset转换到DataFrameds.toDF
  6. Dataset转换到RDDds.rdd

注意:

RDD与DataFrame或者DataSet进行操作,都需要引入隐式转换import spark.implicits._,其中的spark是SparkSession对象的名称!


  
  1. package cn.itcast.sql
  2. import org.apache.spark.SparkContext
  3. import org.apache.spark.rdd.RDD
  4. import org.apache.spark.sql.{DataFrame, Dataset, Row, SparkSession}
  5. /**
  6.  * Author itcast
  7.  * Desc 演示基于RDD/DataFrame/DataSet三者之间的相互转换
  8.  */
  9. object TransformationDemo {
  10.   case class Person(id:Int,name:String,age:Int)
  11.   def main(args: Array[String]): Unit = {
  12.     //1.准备环境-SparkSession
  13.     val spark: SparkSession = SparkSession.builder().appName("SparkSQL").master("local[*]").getOrCreate()
  14.     val sc: SparkContext = spark.sparkContext
  15.     sc.setLogLevel("WARN")
  16.     //2.加载数据
  17.     val lines: RDD[String] = sc.textFile("data/input/person.txt")
  18.     //3.切割
  19.     //val value: RDD[String] = lines.flatMap(_.split(" "))//错误的
  20.     val linesArrayRDD: RDD[Array[String]] = lines.map(_.split(" "))
  21.     //4.将每一行(每一个Array)转为样例类(相当于添加了Schema)
  22.     val personRDD: RDD[Person] = linesArrayRDD.map(arr=>Person(arr(0).toInt,arr(1),arr(2).toInt))
  23.     //5.将RDD转为DataFrame(DF)
  24.     //注意:RDD的API中没有toDF方法,需要导入隐式转换!
  25.     import spark.implicits._
  26.     //转换1:rdd-->df
  27.     val personDF: DataFrame = personRDD.toDF //注意:DataFrame没有泛型
  28.     //转换2:rdd-->ds
  29.     val personDS: Dataset[Person] = personRDD.toDS() //注意:Dataset具有泛型
  30.     //转换3:df-->rdd
  31.     val rdd: RDD[Row] = personDF.rdd //注意:DataFrame没有泛型,也就是不知道里面是Person,所以转为rdd之后统一的使用Row表示里面是很多行
  32.     //转换4:ds-->rdd
  33.     val rdd1: RDD[Person] = personDS.rdd //注意:Dataset具有泛型,所以转为rdd之后还有原来泛型!
  34.     //转换5:ds-->df
  35.     val dataFrame: DataFrame = personDS.toDF()
  36.     //转换5:df-->ds
  37.     val personDS2: Dataset[Person] = personDF.as[Person]
  38.     
  39.     //目前DataFrame和DataSet使用类似,如:也有show/createOrReplaceTempView/select
  40.     personDS.show()
  41.     personDS.createOrReplaceTempView("t_person")
  42.     personDS.select("name").show()
  43.   
  44.   }
  45. }

 

文章来源: lansonli.blog.csdn.net,作者:Lansonli,版权归原作者所有,如需转载,请联系作者。

原文链接:lansonli.blog.csdn.net/article/details/115747643

【版权声明】本文为华为云社区用户转载文章,如果您发现本社区中有涉嫌抄袭的内容,欢迎发送邮件进行举报,并提供相关证据,一经查实,本社区将立刻删除涉嫌侵权内容,举报邮箱: cloudbbs@huaweicloud.com
  • 点赞
  • 收藏
  • 关注作者

评论(0

0/1000
抱歉,系统识别当前为高风险访问,暂不支持该操作

全部回复

上滑加载中

设置昵称

在此一键设置昵称,即可参与社区互动!

*长度不超过10个汉字或20个英文字符,设置后3个月内不可修改。

*长度不超过10个汉字或20个英文字符,设置后3个月内不可修改。