第一章.SparkSQL概述

1.什么是SparkSQL

SparkSQL是Spark用于结构化数据(Structured Data) 处理的Spark模块

$07[SparkSQL(概述_编程)] - 图1

2.为什么要有SparkSQL

$07[SparkSQL(概述_编程)] - 图2

3.SparkSQL原理

$07[SparkSQL(概述_编程)] - 图3

一.什么是DataFrame

  1. 1. DataFrame是一种以RDD为基础的分布式数据集,类似于传统数据库中的二维表格
  2. 2. DataFrame与RDD的主要区别在于,前者带有schema元信息,即DataFrame所表示的二维表数据集每一列都带有名称和类型
  3. 3. SparkSQL性能上比RDD要高,因为SparkSQL了解数据内部结构,从而对藏于DataFrame背后的数据源以及作用于DataFrame之上的变换进行了针对性的优化,最终达到大幅提升运行时效率的目标,反观RDD,由于无从得知所存数据元素的具体内部结构,SparkCore只能在Stage层面进行简单,通用的流水线优化

$07[SparkSQL(概述_编程)] - 图4

执行时间对比:

$07[SparkSQL(概述_编程)] - 图5

二.什么是DataSet

  • DataSet是分布式数据集合
  • DataSet是强类型的,比如可以有DataSet[Car],DataSet[User],具有类型安全检查
  • DataFrame是DataSet的特例, type DataFrame = DataSet[ROW],ROW是一个类型.跟Car,User这些的类型一样,所有的表结构信息都用ROW来表示

三.RDD,DataFrame和DataSet之间的关系

  1. 发展历史
  1. - RDD(Spark1.0) => DataFrame(Spark1.3) => DataSet(Spark1.6)
  2. - 如果同样的数据都给到这三个数据结构,他们分别计算之后,都会给出相同的结果,不同的是他们的执行效率和执行方式,在后期的Spark版本中,DataSet有可能会逐步取代RDD和DataFrame成为唯一的API接口
  1. 三者的共性
  • RDD,DataFrame,DataSet全都是Spark平台下的分布式数据集,为处理超大数据提供便利
  • 三者都有惰性机制,在进行创建,转换,如map方法时,不会立即执行,只有在遇到Action行动算子如foreach时,三者才会遍历计算
  • 三者有许多共同的函数,如filter,排序等
  • 三者都会根据Spark的内存情况自动缓存计算
  • 三者都有分区的概念

四.SparkSQL的优点

  1. 易整合: 无缝的整合了SQL查询和Spark编程

$07[SparkSQL(概述_编程)] - 图6

  1. 统一的数据访问方式:使用相同的方式连接不同的数据源

$07[SparkSQL(概述_编程)] - 图7

  1. 兼容Hive:在已有的仓库上直接运行SQL或者HQL

$07[SparkSQL(概述_编程)] - 图8

  1. 标准的数据连接:通过JDBC或者ODBC来连接

$07[SparkSQL(概述_编程)] - 图9

第二章.SparkSQL编程

1.SparkSession创建

  1. SparkSession介绍
  1. - 在老的版本中,SparkSQL提供两种SQL查询起始点
  2.   - 1.一个叫SparkContext,用于Spark自己提供的SQL查询
  3.   - 2.一个叫HiveContext,用于连接Hive的查询
  4. - SparkSession是Spark最新的SQL查询起始点,实质上是SparkContext和HiveContext的组合,所以SparkContext和HiveContext上可用的API在SparkSession上同样是可以使用的
  5. - SparkSession内部封装了SparkContext,所以计算实际上是由SparkContext完成的,当我们使用spark-shell的时候,spark框架会自动的创建一个名称叫做Spark的SparkSession,就像我们以前可以自动获取到一个sc来表示SparkContext
  1. 添加SparkSQL依赖
  1. <dependency>
  2.       <groupId>org.apache.spark</groupId>
  3.       <artifactId>spark-sql_2.12</artifactId>
  4.       <version>3.0.0</version>
  5.  </dependency>
  1. 编写代码
  1. package com.atguigu.spark.day07
  3. object $01_SparkSession {
  4.   /**
  5.    * SparkSession的创建
  6.    *    1.自己创建内部类对象创建SparkSession
  7.    *      val spark = new sql.SparkSession.Builder().master("local[4]").appName("test").getOrCreate()
  8.    *    2.通过静态方法创建内部类对象创建SparkSession
  9.    *      val spark = SparkSession.builder().master("local[4]").appName("test").getOrCreate()
  10.    */
  12.   def main(args: Array[String]): Unit = {
  13.     //val spark = new sql.SparkSession.Builder().master("local[4]").appName("test").getOrCreate()
  14.     import org.apache.spark.sql.SparkSession
  15.     val spark = SparkSession.builder().master("local[4]").appName("test").getOrCreate()
  16.   }
  17. }

2.DateFrame创建

  1. package com.atguigu.spark.day07
  3. import org.junit.Test
  4. import org.apache.spark.sql.Row
  5. import org.apache.spark.sql.types.{IntegerType, StringType, StructField, StructType}
  7. /**
  8.  * DateFrame的创建方式
  9.  *   1:通过toDF方法创建
  10.  *   2.通过读取文件创建
  11.  *   3.通过createDateFrame API创建
  12.  */
  13. case class Person(id:Int,name:String,age:Int,address:String)
  14. class $02_DateFrameCreate {
  15.   import org.apache.spark.sql.SparkSession
  16.   val spark = SparkSession.builder().master("local[4]").appName("test").getOrCreate()
  18.   /**
  19.    * 1.通过toDF方法创建[toDF是隐式转换方法]
  20.    * 集合.toDF
  21.    * RDD.toDF
  22.    * 无参toDF使用默认的列名[如果集合/RDD中元素类型是样例类,会使用属性名作为默认列名,如果元素类型是元组,会使用_1这种作为默认列名]
  23.    * 工作中如果集合/RDD中元素类型是样例类,一般使用无参toDF方法,如果元素类型是元组,一般使用有参toDF方法重定义默认列名
  24.    */
  25.   @Test
  26.   def createDateFrameByToDF():Unit={
  27.     val list = List(
  28.       Person(1,"lilei1",21,"shenzhen"),
  29.       Person(2,"lilei2",22,"beijing"),
  30.       Person(3,"lilei3",23,"tianjin"),
  31.       Person(4,"lilei4",24,"shanghai"),
  32.     )
  33.     import spark.implicits._
  34.     val df = list.toDF()
  35.     //val df1 = list.toDF("x","y","z","a")
  36.     df.show()
  37.     df.printSchema()
  39.     val rdd = spark.sparkContext.parallelize(List(
  40.       (1,"lilei1",21,"宝安区"),
  41.       (2,"lilei2",22,"龙华区"),
  42.       (3,"lilei3",23,"光明区"),
  43.       (4,"lilei4",24,"福田区")
  44.     ))
  45.     val df2 = rdd.toDF("id", "name", "age", "address")
  46.     df2.show()
  48.   }
  51.   /**
  52.    * 读取文件创建
  53.    */
  54.   @Test
  55.   def createDateFrameReadFile():Unit={
  56.     spark.read.csv("datas/presidential_polls.csv").show()
  58.   }
  60.   /**
  61.    * createDateFrame方法创建
  62.    */
  63.   @Test
  64.   def createDateFrameByApi():Unit={
  65.     val list = List(
  66.       Person(1,"lilei1",21,"shenzhen"),
  67.       Person(2,"lilei2",22,"beijing"),
  68.       Person(3,"lilei3",23,"tianjin"),
  69.       Person(4,"lilei4",24,"shanghai"),
  70.     )
  71.     spark.createDataFrame(list).show()
  73.     val rdd = spark.sparkContext.parallelize(List(
  74.       Row(1,"zhangsan1",21),
  75.       Row(2,"zhangsan2",22),
  76.       Row(3,"zhangsan3",23)
  77.     ))
  78.     val fields = Array[StructField](
  79.       StructField("id1",IntegerType),
  80.       StructField("name1",StringType),
  81.       StructField("age1",IntegerType)
  82.     )
  83.     val schema = StructType(fields)
  84.     val df = spark.createDataFrame(rdd,schema)
  85.     df.show()
  86.   }
  88. }

3.DateSet创建

  1. package com.atguigu.spark.day07
  3. import org.junit.Test
  5. /**
  6.  * DataSet的创建
  7.  *   1.toDS方法创建
  8.  *   2.读取文件创建
  9.  *   3.通过createDateSet API创建
  10.  *
  11.  */
  13. class $03_DataSetCreate {
  14.   import org.apache.spark.sql.SparkSession
  15.   val spark = SparkSession.builder().master("local[4]").appName("test").getOrCreate()
  17.   /**
  18.    * toDS方法创建DataSet
  19.    *   1.集合.toDS
  20.    *   2.RDD.toDS
  21.    *  toDS只有无参的方法,不能重定义列名
  22.    */
  23.   @Test
  24.   def createDateSetByToDS():Unit={
  25.     import spark.implicits._
  26.     val list = List(
  27.       Person(1,"lisi1",21,"shenzhen"),
  28.       Person(2,"lisi2",22,"beijing"),
  29.       Person(3,"lisi3",23,"tianj"),
  30.       Person(4,"lisi4",24,"shanghai")
  31.     )
  32.     val ds = list.toDS()
  33.     ds.printSchema()
  34.     ds.show()
  36.     val list2 = List(
  37.       (1,"lisi1",21,"shenzhen"),
  38.       (2,"lisi2",22,"beijing"),
  39.       (3,"lisi3",23,"tianj"),
  40.       (4,"lisi4",24,"shanghai")
  41.     )
  42.     val rdd1 = spark.sparkContext.parallelize(list2)
  43.     val ds2 = rdd1.toDS()
  44.     ds2.show()
  45.   }
  47.   /**
  48.    * 读取文件创建
  49.    */
  50.   @Test
  51.   def createDateSetByFile():Unit={
  52.     import spark.implicits._
  53.     val ds = spark.read.textFile("datas/wc.txt")
  54.     ds.show()
  55.     val ds2 = ds.flatMap(_.split(" "))
  56.     ds2.show
  58.   }
  60.   /**
  61.    * createDateSet 方法创建
  62.    */
  63.   @Test
  64.   def createDateSetByApi()={
  65.     import spark.implicits._
  66.     val list2 = List(
  67.       (1,"lisi1",21,"shenzhen"),
  68.       (2,"lisi2",22,"beijing"),
  69.       (3,"lisi3",23,"tianj"),
  70.       (4,"lisi4",24,"shanghai")
  71.     )
  72.     val ds = spark.createDataset(list2)
  73.     ds.show()
  74.   }
  75. }

4.DSL编程方式(命令式)

  1. package com.atguigu.spark.day07
  3. import org.junit.Test
  5. class $04_SparkSql {
  7.   import org.apache.spark.sql.SparkSession
  8.   val spark = SparkSession.builder().master("local[4]").appName("test").getOrCreate()
  9.   import spark.implicits._
  11.   /**
  12.    * spark的编程有两种方式:
  13.    *     1.命令式:select,where等算子操作数据
  14.    *     2.声明式:使用SQL语句操作数据
  15.    */
  17.   /**
  18.    * 1.命令式:select,where等算子操作数据
  19.    */
  20.   @Test
  21.   def dsl():Unit={
  22.     val list = List(
  23.       Person(1,"lisi1",21,"shenzhen"),
  24.       Person(2,"lisi2",22,"beijing"),
  25.       Person(2,"lisi2",22,"beijing"),
  26.       Person(2,"lisi2",30,"beijing"),
  27.       Person(3,"lisi3",23,"tianj"),
  28.       Person(4,"lisi4",24,"shanghai"),
  29.       Person(6,"lisi4",35,"shenzhen"),
  30.       Person(7,"lisi4",29,"hangzhou"),
  31.       Person(8,"lisi4",30,"guangzhou")
  32.     )
  33.     val df = list.toDF
  34.     //年龄大于25岁
  35.     df.where("age>25").select("name").show()
  37.     //常用的命令式
  38.     //列裁剪
  39.     /**
  40.      * select 与 selectExpr的区别
  41.      *    1.select默认只能写列名,不能使用函数,不能起别名,但是可以通过隐式转换来使用函数,和起别名
  42.      *    2.selectExpr可以使用sql函数,可以给列起别名
  43.      */
  44.     import org.apache.spark.sql.functions._
  45.     df.select(sum($"age")).show()
  46.     df.selectExpr("sum(age) sum_age").show()
  48.     //去重
  49.     /**
  50.      * distinct:两行数据所有列相同才能去重
  51.      * dropDuplicates:两行数据当指定列相同的时候就会去重
  52.      */
  53.     df.distinct().show()
  54.     df.dropDuplicates("id").show()
  56.     //过滤
  57.     df.where("age>25").show()
  58.     df.filter("age>25").show()
  59.   }
  60. }

5.SQL编程方式(声明式)

  1. /**
  2.    * 声明式:sql语句
  3.    * createOrReplaceTempView:创建临时表,如果表存在则替换[临时表只能在当前sparksession中使用]
  4.    * createOrReplaceGlobalTempView:创建全局表,如果表存在则替换[全局表可以在多个sparksession中使用]
  5.    *    createOrReplaceGlobalTempView创建的表,在使用的时候必须加上global_temp前缀
  6.    */
  8.   @Test
  9.   def sqlMode():Unit={
  10.     val list = List(
  11.       Person(1,"lisi1",21,"shenzhen"),
  12.       Person(2,"lisi2",22,"beijing"),
  13.       Person(2,"lisi2",22,"beijing"),
  14.       Person(2,"lisi2",30,"beijing"),
  15.       Person(3,"lisi3",23,"tianj"),
  16.       Person(4,"lisi4",24,"shanghai"),
  17.       Person(6,"lisi4",35,"shenzhen"),
  18.       Person(7,"lisi4",29,"hangzhou"),
  19.       Person(8,"lisi4",30,"guangzhou")
  20.     )
  21.     val df = list.toDF()
  22.     //将数据集注册成表
  23.     df.createOrReplaceTempView("person")
  24.     spark.sql(
  25.       """
  26.         |select * from person where age>20
  27.         |""".stripMargin
  28.     ).show()
  30.     //临时表只能在当前sparksession中使用
  31.     val spark2 = spark.newSession()
  32.     /*spark2.sql(
  33.       """
  34.         |select * from person where age>20
  35.         |""".stripMargin
  36.     ).show()*/
  37.     df.createOrReplaceGlobalTempView("student")
  38.     spark.sql(
  39.       """
  40.         |select * from global_temp.student where age>20
  41.         |""".stripMargin
  42.     ).show()
  43.     spark2.sql(
  44.       """
  45.         |select * from global_temp.student where age>20
  46.         |""".stripMargin
  47.     ).show()
  49.   }

6.wordcount案例

  1. package com.atguigu.spark.day07
  2. //使用spark编程的方式统计单词个数
  3. object $05_WordCount {
  5.   def main(args: Array[String]): Unit = {
  6.     import org.apache.spark.sql.SparkSession
  7.     val spark = SparkSession.builder().master("local[4]").appName("test").getOrCreate()
  8.     val ds = spark.read.textFile("datas/wc.txt")
  9.     //切割句子得到单词
  10.     import spark.implicits._
  11.     val ds2 = ds.flatMap(_.split(" "))
  12.     ds2.createOrReplaceTempView("wc")
  13.     spark.sql(
  14.       """
  15.         |select value,count(1) num
  16.         |from wc
  17.         |group by value
  18.         |""".stripMargin
  19.     ).show()
  20.   }
  21. }