0
0

spark第四章：SparkSQL基本操作

系列文章目录

spark第一章：环境安装
spark第二章：sparkcore实例
spark第三章：工程化代码
spark第四章：SparkSQL基本操作

文章目录

前言

接下来我们学习SparkSQL他和Hql有些相似。Hql是将操作装换成MR，SparkSQL也是，不过是使用Spark引擎来操作，效率更高一些

一、添加pom

  1. <dependency>
  2.             <groupId>org.apache.spark</groupId>
  3.             <artifactId>spark-sql_2.12</artifactId>
  4.             <version>3.2.3</version>
  5.         </dependency>
  6.         <dependency>
  7.             <groupId>mysql</groupId>
  8.             <artifactId>mysql-connector-java</artifactId>
  9.             <version>5.1.47</version>
  10.         </dependency>
  12.         <dependency>
  13.             <groupId>org.apache.spark</groupId>
  14.             <artifactId>spark-hive_2.12</artifactId>
  15.             <version>3.2.3</version>

以上是这次博客需要的所有依赖，一次性全加上。

二、常用操作

在这里插入图片描述
一共这么多，挨个讲解一下

1.类型转换

SparkSQL中有三种常用的类型，RDD之前说过就不说了。
DataFrame
Spark SQL 的 DataFrame API 允许我们使用 DataFrame 而不用必须去注册临时表或者生成 SQL 表达式。DataFrame API 既有 transformation 操作也有 action 操作。
DSL 语法
DataFrame 提供一个特定领域语言(domain-specific language, DSL)去管理结构化的数据。可以在 Scala, Java, Python 和 R 中使用 DSL，使用 DSL 语法风格不必去创建临时视图了

SparkSql_Basic.scala

  1. package com.atguigu.bigdata.spark.sql
  3. import org.apache.spark.SparkConf
  4. import org.apache.spark.rdd.RDD
  5. import org.apache.spark.sql.{DataFrame, Dataset, Row, SparkSession}
  7. object SparkSql_Basic {
  8.   def main(args: Array[String]): Unit ={
  9.     // 创建SparkSQL的运行环境
  10.     val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("sparkSql")
  11.     val spark: SparkSession = SparkSession.builder().config(sparkConf).getOrCreate()import spark.implicits._
  13. //    val df: DataFrame = spark.read.json("datas/user.json")
  14. //    df.show()
  16.     //DataFrame => SQL
  17. //    df.createOrReplaceTempView("user")
  18. //    spark.sql("select age from user").show()
  20.     //DtaFrame => DSL
  21.     // 在使用DataFrame时，如何涉及到转换操作，需要引入转换规则
  23. //    df.select("age","username").show()
  24. //    df.select($"age"+1).show()
  25. //    df.select('age+1).show()
  27.     // DataSet
  28.     // DataFrame 是特定泛型的DataSet
  29. //    val seq: Seq[Int]= Seq(1, 2, 3, 4)
  30. //    val ds: Dataset[Int]= seq.toDS()
  31. //    ds.show()
  33.     // RDD <=>DataFrame
  34.     val rdd=spark.sparkContext.makeRDD(List((1,"zhangsan",30),(2,"lisi",40)))
  35.     val df: DataFrame = rdd.toDF("id", "name", "age")
  36.     val rowRDD: RDD[Row]= df.rdd
  38.     // DataFrame <=> DatsSet
  39.     val ds: Dataset[User]= df.as[User]
  40.     val df1: DataFrame = ds.toDF()
  42.     // RDD <=> DataSet
  43.     val ds1: Dataset[User]= rdd.map {case(id, name, age)=>{
  44.         User(id, name, age)}}.toDS()
  46.     val userRDD: RDD[User]= ds1.rdd
  48.     // 关闭环境
  49.     spark.close()}case class User(id:Int,name:String,age:Int)}

2.连接mysql

SparkSQL提供了多种数据接口，我们可以通过JDBC连接Mysql数据库，我们先随便在数据库里边写点东西。
在这里插入图片描述
SparkSql_JDBC.scala

  1. package com.atguigu.bigdata.spark.sql
  3. import org.apache.spark.SparkConf
  4. import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}
  6. object SparkSql_JDBC {
  7.   def main(args: Array[String]): Unit ={
  8.     // 创建SparkSQL的运行环境
  9.     val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("sparkSql")
  10.     val spark: SparkSession = SparkSession.builder().config(sparkConf).getOrCreate()import spark.implicits._
  12.     val df: DataFrame = spark.read
  13.       .format("jdbc")
  14.       .option("url", "jdbc:mysql://hadoop102:3306/spark-sql")
  15.       .option("driver", "com.mysql.jdbc.Driver")
  16.       .option("user", "root")
  17.       .option("password", "000000")
  18.       .option("dbtable", "user")
  19.       .option("useSSL","false")
  20.       .load()
  22.     df.show
  24.     df.write
  25.       .format("jdbc")
  26.       .option("url", "jdbc:mysql://hadoop102:3306/spark-sql")
  27.       .option("driver", "com.mysql.jdbc.Driver")
  28.       .option("user", "root")
  29.       .option("password", "000000")
  30.       .option("dbtable", "user1")
  31.       .option("useSSL","false")
  32.       .mode(SaveMode.Append)
  33.       .save()
  35.     // 关闭环境
  36.     spark.close()}}

在这里插入图片描述

在这里插入图片描述

3.UDF函数

这个函数可以对简单的数据进行处理,但是比较局限.
这次我们从json文件读取数据
在这里插入图片描述

  1. {"username":"zhangsan", "age":20}{"username":"lisi", "age":30}{"username":"wangwu", "age":40}

SparkSql_UDF.scala

  1. package com.atguigu.bigdata.spark.sql
  3. import org.apache.spark.SparkConf
  4. import org.apache.spark.rdd.RDD
  5. import org.apache.spark.sql.{DataFrame, Dataset, Row, SparkSession}
  7. object SparkSql_UDF {
  8.   def main(args: Array[String]): Unit ={
  9.     // 创建SparkSQL的运行环境
  10.     val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("sparkSql")
  11.     val spark: SparkSession = SparkSession.builder().config(sparkConf).getOrCreate()import spark.implicits._
  13.     val df: DataFrame = spark.read.json("datas/user.json")
  14.     df.createOrReplaceTempView("user")
  16.     spark.udf.register("prefixName",(name:String)=>{"Name:" + name
  17.     })
  19.     spark.sql("select age ,prefixName(username) from user").show()
  21.     // 关闭环境
  22.     spark.close()}}

在这里插入图片描述

4.UDAF函数

UDAF函数的处理能力就比UDF强大多了,可以完成一些更复杂的操作.
SparkSql_UDAF1.scala

  1. package com.atguigu.bigdata.spark.sql
  3. import org.apache.spark.SparkConf
  4. import org.apache.spark.sql.expressions.Aggregator
  5. import org.apache.spark.sql.types.{DataType, LongType, StructField, StructType}import org.apache.spark.sql.{DataFrame, Encoder, Encoders, Row, SparkSession, functions}
  7. object SparkSql_UDAF1 {
  8.   def main(args: Array[String]): Unit ={
  9.     // 创建SparkSQL的运行环境
  10.     val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("sparkSql")
  11.     val spark: SparkSession = SparkSession.builder().config(sparkConf).getOrCreate()
  13.     val df: DataFrame = spark.read.json("datas/user.json")
  14.     df.createOrReplaceTempView("user")
  16.     //计算平均年龄
  17.     spark.udf.register("ageAvg", functions.udaf(new MyAvgUDAF()))
  19.     spark.sql("select ageAvg(age) from user").show()
  21.     // 关闭环境
  22.     spark.close()}case class Buff( var total:Long,var count:Long)
  24.   class MyAvgUDAF extends Aggregator[Long,Buff,Long]{
  25.     //初始值
  26.     override def zero: Buff ={
  27.       Buff(0L,0L)}
  28.     //更新缓冲区
  29.     override def reduce(buff: Buff, in: Long): Buff ={
  30.       buff.total=buff.total+in
  31.       buff.count=buff.count+1
  32.       buff
  33.     }
  35.     //合并缓冲区
  36.     override def merge(buff1: Buff, buff2: Buff): Buff ={
  37.       buff1.total=buff1.total+buff2.total
  38.       buff1.count=buff1.count+buff2.count
  39.       buff1
  40.     }
  42.     //计算结果
  43.     override def finish(buff: Buff): Long ={
  44.       buff.total/buff.count
  45.     }
  47.     //缓冲区编码操作
  48.     override def bufferEncoder: Encoder[Buff]= Encoders.product
  50.     //输出的编码操作
  51.     override def outputEncoder: Encoder[Long]= Encoders.scalaLong
  52.   }}

在这里插入图片描述
还有一种方法,在Spark3已经不被官方推荐了,所以这里就不叙述了.

5.连接hive

首先我们在集群先,启动Hadoop和Hive
然后将jdbc的jar包放到hive的lib文件中
在这里插入图片描述
这个jar包在安装Hive环境时,使用过.
将虚拟机中的hive配置文件,hive-site.xml导出
在这里插入图片描述
放到idea的resource文件夹中,然后最好吧target文件夹删除,因为idea有可能从target中直接读取之前的数据,从而没有扫描hive-site.xml
在这里插入图片描述
我们就做最简单的查询操作
SparkSql_Hive.scala

  1. package com.atguigu.bigdata.spark.sql
  3. import org.apache.spark.SparkConf
  4. import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}
  6. object SparkSql_Hive {
  7.   def main(args: Array[String]): Unit ={
  9.     val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("sparkSql")
  10.     val spark: SparkSession = SparkSession.builder().enableHiveSupport().config(sparkConf).getOrCreate()
  12.     spark.sql("show tables").show
  14.     // 关闭环境
  15.     spark.close()}}

在这里插入图片描述
如果能查询hive中的数据库,代表成功.

总结

SparkSQL的常用操作基本就这些,至于项目吗,下次专门在写一次吧

标签： spark 大数据 scala
本文转载自: https://blog.csdn.net/weixin_50835854/article/details/129789721
版权归原作者 超哥-- 所有， 如有侵权，请联系我们删除。
登录后发布评论

“spark第四章：SparkSQL基本操作”的评论:

还没有评论
关于作者
...
overfit同步小助手
文章同步
相关阅读

网络安全法-网络运行安全

使用selenium/drissionpage时如何阻止chrome自动跳转http到https

docker desktop 里部署的Open WebUI 管理员密码忘记了的处理方法

在ubuntu20.04中搭建onsite比赛运行环境

利用开源的低代码表单设计器FcDesigner高效管理和渲染复杂表单结构

Kafka学习笔记

【前端】浏览器输入url到页面呈现发生了什么？