- 1【网络】TCP协议——TCP连接相关、TCP连接状态相关、TCP数据传输与控制相关、TCP数据处理和异常、基于TCP应用层协议_tcp协议过程及异常处理
- 2Miniconda、Vscode下载和conda源、pip源设置_miniconda下载
- 3探索未来：ElectronBot——你的桌面级小机器人伙伴
- 4MySQL数据库主从的安装搭建方法_mysql主从搭建教程
- 5【Transformer(04/10) 】 Hugging Face手册-推理管道_hugging face官方文档
- 6一个比 Navicat 更强大、更牛逼的客户端工具
- 7Dropout层到底在干些什么（Pytorch实现）
- 8一张图看懂python-Python 如何将不同数量级的数据画在同一张图上？
- 9最小堆
- 10comfyUi核心节点_prompt outputs failed validation checkpointloaders
SparkSQL
赞
踩
第一章 SparkSQL概述
1.1 SparkSQL是什么
Spark SQL 是Spark用于结构化数据处理的Spark模块
1.2 Hive and SparkSQL
SparkSQL前身是Shark ，给熟悉RDBMS但又不理解MapReduce的技术人员提供快速上手的工具。
Hive 是早期唯一运行在Hadoop上的SQL-on-Hadoop 工具。但那时MapReduce计算过程中大量的中间磁盘落地过程消耗量大量的I/O，降低了运行效率，为了提高SQL-on-Hadoop，大量的工具开始产生，其中突出的是：
- Drill
- Impala
- Shark
Shark是伯克利实验室Spark生态环境的组件之一，是基于Hive所开发的工具，它修改了下图所示的右下角的内存管理、物理计划、执行三个模块，并使之能运行在Spark引擎上。
Shark的出现，使得SQL-on-Hadoop的性能比Hive有了10-100倍的提高。
但是，随着Spark的发展，对于野心勃勃的Spark团队来说，Shark对于Hive的太多依赖（如采用Hive的语法解析器、查询优化器等等），制约了Spark的One Stack Rule Them All的既定方针，制约了Spark各个组件的相互集成，所以提出了SparkSQL项目。SparkSQL抛弃原有Shark的代码，汲取了Shark的一些优点，如内存列存储（In-Memory Columnar Storage）、Hive兼容性等，重新开发了SparkSQL代码； - 数据兼容方面，SparkSQL不但兼容Hive，还可以从RDD ，parquet文件，JSON文件中获取数据，未来版本甚至支持获取RDBMS数据以及cassandra等NOSQL数据；
- 性能优化方面，除了In-Memory ColumnarStorage、byte-code generation等优化技术外、将会引进Cost Model对查询进行动态评估、获取最佳物理计划等等；
- 组件扩展方面，无论是SQL的语法解析器，分析器，还是优化器都可以重新定义，进行扩展。
- 对于开发人员来讲，SparkSQL可以简化RDD的开发，提高开发效率，且执行效率非常快，所以实际工作中，基本上采用的就是SparkSQL。Spark SQL为了简化RDD的开发，提高开发效率，提供了2个编程抽象，类似Spark Core中的RDD
- DataFrame
- DataSet
1.3 SparkSQL特点
1.3.1 易整合
无缝的整合了SQL查询和Spark编程
1.3.2 统一的数据访问
使用相同的方式连接不同的数据源
1.3.3兼容Hive
在已有的仓库是直接运行SQL或者HiveQL
1.3.4 标准数据连接
通过JDBC或者ODBC来连接
1.4 DataFrame是什么
在Spark中，DataFrame是一种以RDD为基础的分布式数据集类似于传统数据库的二维表格。DataFrame与RDD主要区别在于，前者带有schema元信息，即DataFrame所表示的二维表数据集的每一列都带有名称和类型。这使得Spark SQL得以洞察更多的结构信息，从而对藏于DataFrame背后的数据源以及作用于DataFrame之上的变换进行了针对性的优化，最终达到大幅提升运行时效率的目标。反观RDD，由于无从得知所存数据元素的具体内部结构，Spark Core只能在stage层面进行简单、通用的流水线优化。
同时，与Hive类似，DataFrame也支持嵌套数据类型（struct，array，map）。从API易用性来看，DataFrame APi提供的是一套高层的关系操作，比函数式RDD API要更加友好，门槛更低
DataFrame是为数据提供了Schema的视图。可以把它当做数据库中的一张表来对待
DataFrame也是懒执行的，但性能上比RDD要高，主要原因：优化的执行计划，即查询计划通过Spark catalyst optimiser进行优化。比如下面一个例子:
为了说明查询优化，我们来看上图展示的人口数据分析的示例。图中构造了两个DataFrame，将它们join之后又做了一次filter操作。如果原封不动地执行这个执行计划，最终的执行效率是不高的。因为join是一个代价较大的操作，也可能会产生一个较大的数据集。如果我们能将filter下推到 join下方，先对DataFrame进行过滤，再join过滤后的较小的结果集，便可以有效缩短执行时间。而Spark SQL的查询优化器正是这样做的。简而言之，逻辑查询计划优化就是一个利用基于关系代数的等价变换，将高成本的操作替换为低成本操作的过程。
1.5 DataSet是什么
DataSet 是分布式数据集合。DataSet是Spark 1.6中添加的一个新抽象，是DataFrame的一个扩展。它提供了RDD的优势（强类型，使用强大的lambda函数的能力）以及Spark SQL优化执行引擎的优点。DataSet也可以使用功能性的转换（操作map，flatMap，filter等等）。
- DataSet是DataFrame API的一个扩展，是SparkSQL最新的数据抽象
- 用户友好的API风格，既具有类型安全检查也具有DataFrame的查询优化特性；
- 用样例类来对DataSet中定义数据的结构信息，样例类中每个属性的名称直接映射到DataSet中的字段名称；
- DataSet是强类型的，比如可以有DataSet[Car],DataSet[Person].
- DataFrame是DataSet的特例，DataFrame=DataSet[Row],所以可以通过as方法将DataFrame转换为DataSet。Row是一个类型，跟Car、Person这些的类型一样，所有的表结构信息都用Row来表示。获取数据时需要指定顺序
第二章 SparkSQL核心编程
2.1 新的起点
Spark Core中，如果想要执行应用程序，需要首先构建上下文环境对象SparkContext，Spark SQL其实可以理解为对Spark Core的一种封装，不仅仅在模型上进行了封装，上下文环境对象也进行了封装。
在老的版本中，SparkSQL提供两种SQL查询起始点：一个叫SQLContext，用于Spark自己提供的SQL查询；一个叫HiveContext，用于连接Hive的查询。
SparkSession是Spark最新的SQL查询起始点，实质上是SQLContext和HiveContext的组合，所以在SQLContex和HiveContext上可用的API在SparkSession上同样是可以使用的。SparkSession内部封装了SparkContext，所以计算实际上是由sparkContext完成的。当我们使用 spark-shell 的时候, spark框架会自动的创建一个名称叫做spark的SparkSession对象, 就像我们以前可以自动获取到一个sc来表示SparkContext对象一样
2.2 DataFrame
SparkSQL的DataFrameAPI 允许我们使用DataFrame，而不用必须去注册临时表或者生成SQL表达式，DataFrameAPI既有 transformation操作也有action操作。
创建DataFrame
在SparkSQL中，SparkSession是创建DataFrame和执行SQL的入口，创建DataFrame有三种方式：通过Spark的数据源进行创建，从一个存在的RDD进行转换，还可以从HiveTable进行查询返回。
- 从Spark数据源进行创建：
- 查看Spark支持创建文件的数据源格式
- 在Spark的bin/data目录（windows环境）中创建user.json文件
- 读取json文件创建DataFrame
注意：如果从内存中获取数据，spark可以知道数据类型具体是什么。如果是数字，默认作为Int处理；但是从文件中读取的数字，不能确定是什么类型，所以用bigint接收，可以和Long类型转换，但是和Int不能进行转换 - 展示结果 ：show
- 从RDD进行转换
- 从Hive Table进行查询返回
SQL语法
SQL语法风格是指我们查询数据的时候使用SQL语句来查询，这种风格的查询必须要有临时视图或者全局视图来辅助
- 对刚刚读取到的DataFrame创建一个临时表（视图）
- 通过SQL语句实现查询
- 结果展示
注意：普通临时表是Session范围内的，如果想应用范围内有效，可以使用全局临时表。使用全局临时表时需要全路径访问，如：global_temp.people
2.2.3DSL语法
DataFrame提供一个特定领域语言(domain-specific language, DSL)去管理结构化的数据。可以在 Scala, Java, Python 和 R 中使用 DSL，使用 DSL 语法风格不必去创建临时视图了
- 创建一个新的DataFrame
- 查看DataFrame的Schema信息
- 只查看"username"列数据，
- 查看"username"列数据以及"age+1"数据
注意:涉及到运算的时候, 每列都必须使用$, 或者采用引号表达式：单引号+字段名
- 查看"age"大于"20"的数据
- 按照“age”分组
2.2.4RDD转化为DataFrame
在IDEA中开发程序时，如果需要RDD与DF或者DS之间互相操作，那么需要引入 import spark.implicits._
这里的spark不是Scala中的包名，而是创建的sparkSession对象的变量名称，所以必须先创建SparkSession对象再导入。这里的spark对象不能使用var声明，因为Scala只支持val修饰的对象的引入。
spark-shell中无需导入，自动完成此操作。
实际开发中，一般通过样例类将RDD转换为DataFrame
2.2.5DataFrame 转换为RDD
DataFrame其实就是对RDD的封装，所以可以直接获取内部RDD
2.3DataSet
DataSet是具有强类型的数据集合，需要提供对应的类型信息。
2.3.1创建DataSet
- 使用样例类序列创建DataSet
- 使用基本类型的序列创建DataSet（这种很少使用）
2.3.2RDD转为DataSet
SparkSQL能够自动将包含有case类的RDD转成DataSet，case类定义了table的结构，case类属性通过反射变成了表的列名。Case类可以包含诸如Seq或者Array等复杂结构
2.3.3DataSet转换为RDD
DataSet其实也是对RDD的封装，所以可以直接获取内部的RDD
2.4 DataFrame和DataSet转换
DataFrame其实是DataSet的特例，所以他们可以互相转换
- DataFrame转换为DataSet
- DataSet转换为DataFrame
2.5.2 三者的区别
- RDD
- RDD一般和spark mllib同时使用
- RDD不支持sparksql操作
- DataFrame
- 与RDD不同，DataFrame每一行类型都固定位Row，每一列的值没办法直接访问，只有通过解析
- DataFrame与DataSet一般不与 spark mllib 同时使用
- DataFrame与DataSet均支持 SparkSQL 的操作，比如select，groupby之类，还能注册临时表/视窗，进行 sql 语句操作
- DataFrame与DataSet支持一些特别方便的保存方式，比如保存成csv，可以带上表头，这样每一列的字段名一目了然
- DataSet
- Dataset和DataFrame拥有完全相同的成员函数，区别只是每一行的数据类型不同。 DataFrame其实就是DataSet的一个特例 type DataFrame = Dataset[Row]
2.5.3 三者的相互转换
2.6 IDEA开发SparkSQL
2.6.1 添加依赖
<groupId>org.apache.spark</groupId>
<artifactId>spark-sql_2.12</artifactId>
<version>3.0.0</version>
</dependency>
- 1
- 2
- 3
- 4
2.6.2 代码实现
import org.apache.spark.sql.{DataFrame, Dataset, Row, SparkSession} import org.apache.spark.SparkConf import org.apache.spark.rdd.RDD // 定义样例类 case class User(id:Int,name:String,age:Int) object SparkSQL01_Demo { def main(args: Array[String]): Unit = { //创建上下文环境配置对象 val conf = new SparkConf().setMaster("local[*]").setAppName("SparkSQL") //创建SparkSession对象 val ss: SparkSession = SparkSession.builder().config(conf).getOrCreate() //RDD=>DataFrame=>DataSet转换需要引入隐式转换规则，否则无法转换 //spark不是包名，是上下文环境对象名 import ss.implicits._ //读取json文件 创建DataFrame val df: DataFrame = ss.read.json("F:/worksapce_java/bigdata/SparkSQL/data/person.json") //df.show //SQL风格 df.createOrReplaceTempView("user") //ss.sql("select * from user").show //DSL 风格语法 //df.select("username","age").show() //RDD与DataFrame，DataSet互换 val rdd1: RDD[(Int, String, Int)] = ss.sparkContext.makeRDD(List((1,"zhangsan",30),(2,"lisi",28),(3,"wangwu",20))) // RDD => DataFrame val df1: DataFrame = rdd1.toDF("id","name","age") //df1.show() // DataFrame => DataSet val ds1: Dataset[User] = df1.as[User] //ds1.show() // DataSet=>DataFrame val df2: DataFrame = ds1.toDF() //df2.show() //DataFrame => RDD 返回的RDD类型为Row，里面提供的getXXX方法可以获取字段值，类似jdbc处理结果集，但是索引从0开始 val rdd2: RDD[Row] = df2.rdd //rdd2.foreach(a=>println(a.getString(1))) //RDD=>DataSet val ds2: Dataset[User] = rdd1.map { case (id, name, age) => User(id, name, age) }.toDS() ds2.show() //DataSet => RDD val rdd3: RDD[User] = ds2.rdd ss.stop() } }
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
- 44
- 45
- 46
- 47
2.7 自定义函数
用户可以通过spark.udf功能添加自定义函数，实现自定义功能。
2.7.1 UDF
val df: DataFrame = ss.read.json("F:/worksapce_java/bigdata/SparkSQL/data/person.json")
//注册UDF
ss.udf.register("addName",(x:String)=>"Name："+x)
//创建临时表
df.createOrReplaceTempView("people")
//应用UDF
ss.sql("select addName(name),age from people").show
- 1
- 2
- 3
- 4
- 5
- 6
- 7
2.7.2 UDAF
强类型的Dataset和弱类型的DataFrame都提供了相关的聚合函数， 如 count()，countDistinct()，avg()，max()，min()。除此之外，用户可以设定自己的自定义聚合函数。通过继承UserDefinedAggregateFunction来实现用户自定义弱类型聚合函数。从Spark3.0版本后，UserDefinedAggregateFunction已经不推荐使用了。可以统一采用强类型聚合函数Aggregator
接下来通过一个需求讲述什么是UDAF：需求：计算平均年龄
- 实现方式-RDD
val conf: SparkConf = new SparkConf().setAppName("app").setMaster("local[*]")
val sc: SparkContext = new SparkContext(conf)
val res: (Int, Int) = sc.makeRDD(List(("zhangsan", 20), ("lisi", 30), ("wangw", 40))).map {
case (name, age) => {
(age, 1)
}
}.reduce {
(t1, t2) => {
(t1._1 + t2._1, t1._2 + t2._2)
}
}
println(res._1/res._2)
// 关闭连接
sc.stop()
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 实现方式-累加器
class MyAC extends AccumulatorV2[Int,Int]{ var sum:Int = 0 var count:Int = 0 override def isZero: Boolean = { return sum ==0 && count == 0 } override def copy(): AccumulatorV2[Int, Int] = { val newMyAc = new MyAC newMyAc.sum = this.sum newMyAc.count = this.count newMyAc } override def reset(): Unit = { sum =0 count = 0 } override def add(v: Int): Unit = { sum += v count += 1 } override def merge(other: AccumulatorV2[Int, Int]): Unit = { other match { case o:MyAC=>{ sum += o.sum count += o.count } case _=> } } override def value: Int = sum/count }
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 实现方式 -UDAF-弱类型
/* 定义类继承UserDefinedAggregateFunction，并重写其中方法 */ class MyAveragUDAF extends UserDefinedAggregateFunction { // 聚合函数输入参数的数据类型 def inputSchema: StructType = StructType(Array(StructField("age",IntegerType))) // 聚合函数缓冲区中值的数据类型(age,count) def bufferSchema: StructType = { StructType(Array(StructField("sum",LongType),StructField("count",LongType))) } // 函数返回值的数据类型 def dataType: DataType = DoubleType // 稳定性：对于相同的输入是否一直返回相同的输出。 def deterministic: Boolean = true // 函数缓冲区初始化 def initialize(buffer: MutableAggregationBuffer): Unit = { // 存年龄的总和 buffer(0) = 0L // 存年龄的个数 buffer(1) = 0L } // 更新缓冲区中的数据 def update(buffer: MutableAggregationBuffer,input: Row): Unit = { if (!input.isNullAt(0)) { buffer(0) = buffer.getLong(0) + input.getInt(0) buffer(1) = buffer.getLong(1) + 1 } } // 合并缓冲区 def merge(buffer1: MutableAggregationBuffer,buffer2: Row): Unit = { buffer1(0) = buffer1.getLong(0) + buffer2.getLong(0) buffer1(1) = buffer1.getLong(1) + buffer2.getLong(1) } // 计算最终结果 def evaluate(buffer: Row): Double = buffer.getLong(0).toDouble / buffer.getLong(1) } 。。。 //创建聚合函数 var myAverage = new MyAveragUDAF //在spark中注册聚合函数 spark.udf.register("avgAge",myAverage) spark.sql("select avgAge(age) from user").show()
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
- 44
- 45
- 46
- 47
- 48
- 49
- 50
- 51
- 52
- 53
- 54
- 55
- 实现方式 - UDAF - 强类型
// TODO 创建UDAF函数 val udaf = new MyAvgAgeUDAF // TODO 注册到SparkSQL中 spark.udf.register("avgAge", functions.udaf(udaf)) // TODO 在SQL中使用聚合函数 // 定义用户的自定义聚合函数 spark.sql("select avgAge(age) from user").show // ************************************************** case class Buff( var sum:Long, var cnt:Long ) // totalage, count class MyAvgAgeUDAF extends Aggregator[Long, Buff, Double]{ override def zero: Buff = Buff(0,0) override def reduce(b: Buff, a: Long): Buff = { b.sum += a b.cnt += 1 b } override def merge(b1: Buff, b2: Buff): Buff = { b1.sum += b2.sum b1.cnt += b2.cnt b1 } override def finish(reduction: Buff): Double = { reduction.sum.toDouble/reduction.cnt } override def bufferEncoder: Encoder[Buff] = Encoders.product override def outputEncoder: Encoder[Double] = Encoders.scalaDouble }
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
2.8 数据加载和保存
2.8.1 通用的加载和保存方式
SparkSQL提供了通用的保存数据和数据加载的方式，这里的通用指的是使用相同的API，根据不同的参数，读取和保存不同格式的数据，SparkSQL默认读取和保存的文件格式为parquet
- 加载数据
spark.read.load是加载数据的通用方法，如果读取不同格式的数据，可以对不同的数据格式进行设定：
spark.read.format(“…”)[option(“…”)].load(“…”)
format(“…”):指定加载的数据类型，包括"csv"、“jdbc”、“json”、“orc”、“parquet"和"textFile”。
load(“…”)：在"csv"、“jdbc”、“json”、“orc”、“parquet"和"textFile"格式下需要传入加载数据的路径
option(”…")：在"jdbc"格式下需要传入JDBC相应参数，url、user、password和dbtable
我们前面都是使用read API 先把文件加载到 DataFrame然后再查询，其实，我们也可以直接在文件上进行查询: 文件格式.文件路径
spark.sql("select * from json.`/opt/module/data/user.json`").show
- 1
- 保存数据
对数据类型设置的方法和读取文件一致
保存操作可以使用 SaveMode, 用来指明如何处理数据，使用mode()方法来设置。
有一点很重要: 这些 SaveMode 都是没有加锁的, 也不是原子操作
强调一下，保存文件并不是只生成一个json文件，而是生成一个命名文件夹,追加时只是往文件夹追加文件
2.8.2 Parquet
Spark SQL的默认数据源为Parquet格式。Parquet是一种能够有效存储嵌套数据的列式存储格式。
数据源为Parquet文件时，Spark SQL可以方便的执行所有的操作，不需要使用format。修改配置项spark.sql.sources.default，可修改默认数据源格式。
2.8.3 JSON
Spark SQL 能够自动推测JSON数据集的结构，并将它加载为一个Dataset[Row]. 可以通过SparkSession.read.json()去加载JSON 文件。
**注意：Spark读取的JSON文件不是传统的JSON文件，每一行都应该是一个JSON串。**格式如下：
{"name":"Michael"}
{"name":"Andy"， "age":30}
[{"name":"Justin"， "age":19},{"name":"Justin"， "age":19}]
- 1
- 2
- 3
2.8.4 CSV
Spark SQL可以配置CSV文件的列表信息，读取CSV文件,CSV文件的第一行设置为数据列
spark.read.format("csv").option("sep", ";").option("inferSchema", "true").option("header", "true").load("data/user.csv")
- 1
2.8.5 MySQL
Spark SQL可以通过JDBC从关系型数据库中读取数据的方式创建DataFrame，通过对DataFrame一系列的计算后，还可以将数据再写回关系型数据库中。如果使用spark-shell操作，可在启动shell时指定相关的数据库驱动路径或者将相关的数据库驱动放到spark的类路径下。
bin/spark-shell
--jars mysql-connector-java-8.0.26-bin.jar
- 1
- 2
这里只学习Idea中通过JDBC对Mysql进行操作
- 导入依赖
<dependency>
<groupId>mysql</groupId>
<artifactId>mysql-connector-java</artifactId>
<version>8.0.26</version>
</dependency>
- 1
- 2
- 3
- 4
- 5
- 读取数据
def main(args: Array[String]): Unit = { val conf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("JDBC") val ss: SparkSession = SparkSession.builder().config(conf).getOrCreate() import ss.implicits._ //方式一 : 通用的load方法读取 ss.read.format("jdbc") .option("url", "jdbc:mysql://192.168.23.100:3306/spark_sql") .option("driver", "com.mysql.jdbc.Driver") .option("user","root") .option("password","123456") .option("dbtable","user") .load().show() //方式二 : 通用的load方法读取 参数另一种形式 ss.read.format("jdbc") .options(Map("url"->"jdbc:mysql://192.168.23.100:3306/spark_sql?user=root&password=123456", "dbtable"->"user","driver"->"com.mysql.jdbc.Driver")) .load().show() //方式三 : 使用jdbc方法读取 val prop = new Properties() prop.setProperty("user","root") prop.setProperty("password", "123456") val df: DataFrame = ss.read.jdbc("jdbc:mysql://192.168.23.100:3306/spark_sql","user",prop) df.show() ss.stop() }
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 写入数据
case class User2(name: String, age: Long) def main(args: Array[String]): Unit = { val conf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("JDBC") val ss: SparkSession = SparkSession.builder().config(conf).getOrCreate() import ss.implicits._ val rdd: RDD[User2] = ss.sparkContext.makeRDD(List(User2("王", 25), User2("赵", 26))) val ds: Dataset[User2] = rdd.toDS //方式一 : 通用的load方法读取 ds.write.format("jdbc") .option("url", "jdbc:mysql://192.168.23.100:3306/spark_sql") .option("driver", "com.mysql.jdbc.Driver") .option("user", "root") .option("charset", "utf8") .option("password", "123456") .option("dbtable", "user") .mode(SaveMode.Append).save() //方式二 ： 通过jdbc方法 // val props: Properties = new Properties() // props.setProperty("user", "root") // props.setProperty("password", "123456") // ds.write.mode(SaveMode.Append).jdbc("jdbc:mysql://192.168.23.100:3306/spark_sql", "user", props) ss.stop() }
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
2.8.6 Hive
Apache Hive 是 Hadoop 上的 SQL 引擎，Spark SQL编译时可以包含 Hive 支持，也可以不包含。包含 Hive 支持的 Spark SQL 可以支持 Hive 表访问、UDF (用户自定义函数)以及 Hive 查询语言(HiveQL/HQL)等。需要强调的一点是，如果要在 Spark SQL 中包含Hive 的库，并不需要事先安装 Hive。一般来说，最好还是在编译Spark SQL时引入Hive支持，这样就可以使用这些特性了。如果你下载的是二进制版本的 Spark，它应该已经在编译时添加了 Hive 支持。
若要把 Spark SQL 连接到一个部署好的 Hive 上，你必须把 hive-site.xml 复制到 Spark的配置文件目录中($SPARK_HOME/conf)。即使没有部署好 Hive，Spark SQL 也可以运行。 需要注意的是，如果你没有部署好Hive，Spark SQL 会在当前的工作目录中创建出自己的 Hive 元数据仓库，叫作 metastore_db。此外，如果你尝试使用 HiveQL 中的 CREATE TABLE (并非 CREATE EXTERNAL TABLE)语句来创建表，这些表会被放在你默认的文件系统中的 /user/hive/warehouse 目录中(如果你的 classpath 中有配好的 hdfs-site.xml，默认的文件系统就是 HDFS，否则就是本地文件系统)。
spark-shell默认是Hive支持的；代码中是默认不支持的，需要手动指定（加一个参数即可）。
- 内嵌的HIVE 如果使用Spark 内嵌的HIve 什么都不要做，直接使用即可。Hive元素局保存在derby中， 默认仓库地址:$SPARK_HOME/spark-warehouse
（在实际使用中, 几乎没有任何人会使用内置的 Hive） - 外部的HIVE
如果在shell想连接外部已经部署好的Hive，需要通过以下几个步骤：
1 . Spark要接管Hive需要把hive-site.xml拷贝到conf/目录下 - 把Mysql的驱动copy到jars/目录下
- 如果访问不到hdfs，则需要把core-site.xml和hdfs-site.xml拷贝到conf/目录下
- 重启spark-shell
如果使用Idea操作Hive：
- 导入依赖
<dependency> <groupId>org.apache.spark</groupId> <artifactId>spark-hive_2.12</artifactId> <version>3.0.0</version> </dependency> <dependency> <groupId>org.apache.hive</groupId> <artifactId>hive-exec</artifactId> <version>1.2.1</version> </dependency> <dependency> <groupId>mysql</groupId> <artifactId>mysql-connector-java</artifactId> <version>5.1.27</version> </dependency>
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 将hive-site.xml文件拷贝到项目的resources目录中，代码实现
case class User2(name: String, age: Long)
def main(args: Array[String]): Unit = {
val ss: SparkSession = SparkSession.builder().master("local[*]").appName("Hive").enableHiveSupport().getOrCreate()
import ss.implicits._
ss.sql("show databases ").show()
ss.stop()
}
- 1
- 2
- 3
- 4
- 5
- 6
- 7
