啥叫大数据_SQL数据类型

一、Spark SQL概述

Spark SQL是Spark用来处理结构化数据的一个模块，它提供了两个编程抽象叫做DataFrame和DataSet并且作为分布式SQL查询引擎的作用，其实也是对RDD的再封装。大数据Hadoop之——计算引擎Spark，官方文档：https://spark.apache.org/sql/

二、SparkSQL版本

1）SparkSQL的演变之路

• 1.0以前： Shark（入口：SQLContext和HiveContext）

  1. SQLContext：主要DataFrame的构建以及DataFrame的执行，SQLContext指的是spark中SQL模块的程序入口。
  2. HiveContext：是SQLContext的子类，专门用于与Hive的集成，比如读取Hive的元数据，数据存储到Hive表、Hive的窗口分析函数等。

• 1.1.x开始：SparkSQL(只是测试性的)

• 1.3.x: SparkSQL(正式版本)+Dataframe

• 1.5.x: SparkSQL 钨丝计划

• 1.6.x： SparkSQL+DataFrame+DataSet(测试版本)

• 2.x:

  1. 入口：SparkSession（spark应用程序的一个整体入口），合并了SQLContext和HiveContext
  2. SparkSQL+DataFrame+DataSet(正式版本)
  3. Spark Streaming-》Structured Streaming(DataSet)

2）shark与SparkSQL对比

• shark
  1. 执行计划优化完全依赖于Hive，不方便添加新的优化策略；
  2. Spark是线程级并行，而MapReduce是进程级并行。
  3. Spark在兼容Hive的实现上存在线程安全问题，导致Shark
     不得不使用另外一套独立维护的打了补丁的Hive源码分支；
• Spark SQL
  1. 作为Spark生态的一员继续发展，而不再受限于Hive，
  2. 只是兼容Hive；Hive on Spark作为Hive的底层引擎之一
  3. Hive可以采用Map-Reduce、Tez、Spark等引擎

3）SparkSession

• SparkSession是Spark 2.0引如的新概念。SparkSession为用户提供了统一的切入点，来让用户学习spark的各项功能。
• 在spark的早期版本中，SparkContext是spark的主要切入点，由于RDD是主要的API，我们通过sparkcontext来创建和操作RDD。对于每个其他的API，我们需要使用不同的context。

【例如】对于Streming，我们需要使用StreamingContext；对于sql，使用sqlContext；对于Hive，使用hiveContext。但是随着DataSet和DataFrame的API逐渐成为标准的API，就需要为他们建立接入点。所以在spark2.0中，引入SparkSession作为DataSet和DataFrame API的切入点，SparkSession封装了SparkConf、SparkContext和SQLContext。为了向后兼容，SQLContext和HiveContext也被保存下来。

• SparkSession实质上是SQLContext和HiveContext的组合（未来可能还会加上StreamingContext），所以在SQLContext和HiveContext上可用的API在SparkSession上同样是可以使用的。SparkSession内部封装了sparkContext，所以计算实际上是由sparkContext完成的，在spark 2.x中不推荐使用SparkContext对象读取数据，而是推荐SparkSession。

三、RDD、DataFrames和DataSet

1）三者关联关系

DataFrame 和 DataSet 是 Spark SQL 提供的基于 RDD 的结构化数据抽象。它既有 RDD 不可变、分区、存储依赖关系等特性，又拥有类似于关系型数据库的结构化信息。所以，基于 DataFrame 和 DataSet API 开发出的程序会被自动优化，使得开发人员不需要操作底层的 RDD API 来进行手动优化，大大提升开发效率。但是 RDD API 对于非结构化的数据处理有独特的优势，比如文本流数据，而且更方便我们做底层的操作。

1）RDD

RDD（Resilient Distributed Dataset）叫做弹性分布式数据集，是Spark中最基本的数据抽象，它代表一个不可变、可分区、里面的元素可并行计算的集合。RDD具有数据流模型的特点：自动容错、位置感知性调度和可伸缩性。RDD允许用户在执行多个查询时显式地将工作集缓存在内存中，后续的查询能够重用工作集，这极大地提升了查询速度。

1、核心概念

• 一组分片（Partition）：即数据集的基本组成单位。对于RDD来说，每个分片都会被一个计算任务处理，并决定并行计算的粒度。用户可以在创建RDD时指定RDD的分片个数，如果没有指定，那么就会采用默认值。默认值就是程序所分配到的CPU Core的数目。

• 一个计算每个分区的函数。Spark中RDD的计算是以分片为单位的，每个RDD都会实现compute函数以达到这个目的。compute函数会对迭代器进行复合，不需要保存每次计算的结果。

• RDD之间的依赖关系：RDD的每次转换都会生成一个新的RDD，所以RDD之间就会形成类似于流水线一样的前后依赖关系。在部分分区数据丢失时，Spark可以通过这个依赖关系重新计算丢失的分区数据，而不是对RDD的所有分区进行重新计算。

• 一个Partitioner：即RDD的分片函数。当前Spark中实现了两种类型的分片函数，一个是基于哈希的HashPartitioner，另外一个是基于范围的RangePartitioner。只有对于于key-value的RDD，才会有Partitioner，非key-value的RDD的Parititioner的值是None。Partitioner函数不但决定了RDD本身的分片数量，也决定了parent RDD Shuffle输出时的分片数量。

• 一个列表：存储存取每个Partition的优先位置（preferred location）。对于一个HDFS文件来说，这个列表保存的就是每个Partition所在的块的位置。按照“移动数据不如移动计算”的理念，Spark在进行任务调度的时候，会尽可能地将计算任务分配到其所要处理数据块的存储位置。

2、RDD简单操作

启动spark-shell，其实spark-shell低层也是调用spark-submit，首先需要配置好，当然也可以写在命令行，但是不推荐。配置如下，仅供参考（这里使用yarn模式）：

$ cat spark-defaults.conf

启动spark-shell（下面会详解讲解）

$ spark-shell

【问题】发现有个WARN：WARN yarn.Client: Neither spark.yarn.jars nor spark.yarn.archive is set, falling back to uploading libraries under SPARK_HOME.
【原因】是因为Spark提交任务到yarn集群，需要上传相关spark的jar包到HDFS。
【解决】 提前上传到HDFS集群，并且在Spark配置文件指定文件路径，就可以避免每次提交任务到Yarn都需要重复上传文件。下面是解决的具体操作步骤：

### 打包jars，jar相关的参数说明
#-c  创建一个jar包
# -t 显示jar中的内容列表
#-x 解压jar包
#-u 添加文件到jar包中
#-f 指定jar包的文件名
#-v  生成详细的报造，并输出至标准设备
#-m 指定manifest.mf文件.(manifest.mf文件中可以对jar包及其中的内容作一些一设置)
#-0 产生jar包时不对其中的内容进行压缩处理
#-M 不产生所有文件的清单文件(Manifest.mf)。这个参数与忽略掉-m参数的设置
#-i    为指定的jar文件创建索引文件
#-C 表示转到相应的目录下执行jar命令,相当于cd到那个目录，然后不带-C执行jar命令
$ cd /opt/bigdata/hadoop/server/spark-3.2.0-bin-hadoop3.2
$ jar cv0f spark-libs.jar -C ./jars/ .
$ ll
### 在hdfs上创建存放jar包目录
$ hdfs dfs -mkdir -p /spark/jars
## 上传jars到HDFS
$ hdfs dfs -put spark-libs.jar /spark/jars/
## 增加配置spark-defaults.conf 
spark.yarn.archive=hdfs:///spark/jars/spark-libs.jar

然后再启动spark-shell

在Spark Shell中，有一个专有的SparkContext已经为您创建好了，变量名叫做sc，自己创建的SparkContext将无法工作。

$ spark-shell

### 由一个已经存在的Scala集合创建。
val array = Array(1,2,3,4,5)
# spark使用parallelize方法创建RDD
val rdd = sc.parallelize(array)

这里只是简单的创建RDD操作，后面会有更多RDD相关的演示操作。

3、RDD API

Spark支持两个类型（算子）操作：Transformation和Action

1）Transformation

主要做的是就是将一个已有的RDD生成另外一个RDD。Transformation具有lazy特性(延迟加载)。Transformation算子的代码不会真正被执行。只有当我们的程序里面遇到一个action算子的时候，代码才会真正的被执行。这种设计让Spark更加有效率地运行。

常用的Transformation：

2）Action

触发代码的运行，我们一段spark代码里面至少需要有一个action操作。

常用的Action:

4、实战操作

1、针对各个元素的转化操作

我们最常用的转化操作应该是map() 和filter()，转化操作map() 接收一个函数，把这个函数用于RDD 中的每个元素，将函数的返回结果作为结果RDD 中对应元素的值。而转化操作filter() 则接收一个函数，并将RDD 中满足该函数的元素放入新的RDD 中返回。

让我们看一个简单的例子，用map() 对RDD 中的所有数求平方

# 通过parallelize创建RDD对象
val input = sc.parallelize(List(1, 2, 3, 4))
val result = input.map(x => x * x)
println(result.collect().mkString(","))

2、对一个数据为{1，2，3，3}的RDD进行基本的RDD转化操作（去重）

var rdd = sc.parallelize(List(1,2,3,3))
rdd.distinct().collect().mkString(",")

3、对数据分别为{1，2，3}和{3，4，5}的RDD进行针对两个RDD的转化操作

var rdd = sc.parallelize(List(1,2,3))
var other = sc.parallelize(List(3,4,5))
# 生成一个包含两个RDD中所有元素的RDD
rdd.union(other).collect().mkString(",")
# 求两个RDD共同的元素RDD
rdd.intersection(other).collect().mkString(",")

4、行动操作

行动操作reduce()，它接收一个函数作为参数，这个函数要操作两个RDD 的元素类型的数据并返回一个同样类型的新元素。一个简单的例子就是函数+，可以用它来对我们的RDD 进行累加。使用reduce()，可以很方便地计算出RDD中所有元素的总和、元素的个数，以及其他类型的聚合操作。

var rdd = sc.parallelize(List(1,2,3,4,5,6,7))
# 求和
var sum = rdd.reduce((x, y) => x + y)
# 求元素个数
var sum = rdd.count()

# 聚合操作
var rdd = sc.parallelize(List(1,2,3,4,5,6,7))
var result = rdd.aggregate((0,0))((acc,value) => (acc._1 + value,acc._2 + 1),(acc1,acc2) => (acc1._1 + acc2._1 , acc1._2 + acc2._2))
var avg = result._1/result._2.toDouble

这里只是演示几个简单的示例，更多RDD的操作，可以参考官方文档学习哦。

2）DataFrames

在Spark中，DataFrame提供了一个领域特定语言(DSL)和SQL来操作结构化数据，DataFrame是一种以RDD为基础的分布式数据集，类似于传统数据库中的二维表格。

• RDD，由于无从得知所存数据元素的具体内部结构，Spark Core只能在stage层面进行简单、通用的流水线优化。
• DataFrame底层是以RDD为基础的分布式数据集，和RDD的主要区别的是：RDD中没有schema信息，而DataFrame中数据每一行都包含schema。DataFrame = RDD + shcema

1、DSL风格语法操作

1）DataFrame创建

创建DataFrame的两种基本方式：

• 已存在的RDD调用toDF()方法转换得到DataFrame。
• 通过Spark读取数据源直接创建DataFrame。

直接创建DataFarme对象

若使用SparkSession方式创建DataFrame，可以使用spark.read从不同类型的文件中加载数据创建DataFrame。spark.read的具体操作，如下所示。

1、在本地创建一个person.txt文本文档，用于读取：运行spark-shell：

# person.txt，Name,Age,Height
p1_name,18,165
p2_name,19,170
p3_name,20,188
p4_name,21,190

# 启动spark shell，默认会创建一个spark名称的spark session对象
$ spark-shell
# 定义变量，【注意】所有节点都得创建这个person文件，要不然调度没有这个文件的机器会报错
var inputFile = "file:///opt/bigdata/hadoop/server/spark-3.2.0-bin-hadoop3.2/test/person.txt"
# 读取本地文件
val personDF = spark.read.text("file:///opt/bigdata/hadoop/server/spark-3.2.0-bin-hadoop3.2/test/person.txt")
val personDF = spark.read.text(inputFile)
# 显示
personDF.show()
# 将文件put到hdfs上
# 读取hdfs文件（推荐）
val psersonDF = spark.read.text("hdfs:///person.txt")

2、有RDD转换成DataFrame

# 读取文本文档，按逗号分割开来
val lineRDD = sc.textFile("hdfs:///person.txt").map(_.split(","))
case class Person(name:String, age:Int, height:Int)
# 按照样式类对RDD数据进行分割成map
val personRDD = lineRDD.map(x => Person(x(0).toString, x(1).toInt, x(2).toInt))
# 把RDD数据类型转成DataFarme
val personDF = personRDD.toDF()
# 查看这个表
personDF.show()
# 查看Schema数据
personDF.printSchema()
# 查看列
personDF.select(personDF.col("name")).show
# 过滤年龄小于25的
personDF.filter(col("age") >= 25).show

这里提供常用的spark dataframe方法：

这里已经列出了很多常用方法了，基本上涵盖了大部分操作，当然也可以参考官方文档

2、SQL风格语法操作

DataFrame的一个强大之处就是我们可以将它看作是一个关系型数据表，然后可以通过在程序中使用spark.sql() 来执行SQL查询，结果将作为一个DataFrame返回。因为spark session包含了Hive Context，所以spark.sql() 会自动启动连接hive，默认模式就是hive里的local模式（内嵌derby）

启动spark-shell

$ spark-shell

会在执行spark-shell当前目录下生成两个文件：derby.log，metastore_db

接下来就可以happy的写sql了，这里就演示几个命令，跟之前的hive一样，把sql语句放在spark.sql()方法里执行即可，不清楚hive sql的可以参考我之前的文章：大数据Hadoop之——数据仓库Hive

# 有个默认default库
$ spark.sql("show databases").show
# 默认当前库是default
$ spark.sql("show tables").show

通过spark-sql启动spark shell

操作就更像sql语法了，已经跟hive差不多了。接下来演示几个命令，大家就很清楚了。

$ spark-sql
show databases;
create database test007

同样也会在当前目录下自动创建两个文件：derby.log，metastore_db

3）DataSet

DataSet是分布式的数据集合，Dataset提供了强类型支持，也是在RDD的每行数据加了类型约束。DataSet是在Spark1.6中添加的新的接口。它集中了RDD的优点（强类型和可以用强大lambda函数）以及使用了Spark SQL优化的执行引擎。DataSet可以通过JVM的对象进行构建，可以用函数式的转换（map/flatmap/filter）进行多种操作。

1、通过spark.createDataset通过集合进行创建dataSet

val ds1 = spark.createDataset(1 to 10)
ds1.show

2、从已经存在的rdd当中构建dataSet

官方文档

val ds2 = spark.createDataset(sc.textFile("hdfs:////person.txt"))

3、通过样例类配合创建DataSet

case class Person(name:String,age:Int)
val personDataList = List(Person("zhangsan",18),Person("lisi",28))
val personDS = personDataList.toDS
personDS.show

4、通过DataFrame转化生成
Music.json文件内容如下：

{“name”:”上海滩”,”singer”:”叶丽仪”,”album”:”香港电视剧主题歌”,”path”:”mp3/shanghaitan.mp3″}
{“name”:”一生何求”,”singer”:”陈百强”,”album”:”香港电视剧主题歌”,”path”:”mp3/shanghaitan.mp3″}
{“name”:”红日”,”singer”:”李克勤”,”album”:”怀旧专辑”,”path”:”mp3/shanghaitan.mp3″}
{“name”:”爱如潮水”,”singer”:”张信哲”,”album”:”怀旧专辑”,”path”:”mp3/airucaoshun.mp3″}
{“name”:”红茶馆”,”singer”:”陈惠嫻”,”album”:”怀旧专辑”,”path”:”mp3/redteabar.mp3″}

case class Music(name:String,singer:String,album:String,path:String)
# 注意把test.json传到hdfs上
val jsonDF = spark.read.json("hdfs:///Music.json")
val jsonDS = jsonDF.as[Music]
jsonDS.show

RDD，DataFrame，DataSet互相转化

四、RDD、DataFrame和DataSet的共性与区别

• RDD[Person]：以Person为类型参数，但不了解 其内部结构。

• DataFrame：提供了详细的结构信息schema（结构）列的名称和类型。这样看起来就像一张表了

• DataSet[Person]：不光有schema（结构）信息，还有类型信息

1）共性

• 三者都是spark平台下的分布式弹性数据集，为处理超大型数据提供便利
• 三者都有惰性机制。在创建时、转换时(如map)不会立即执行，只有在遇到action算子的时候(比如foreach)，才开始进行触发计算。极端情况下，如果代码中只有创建、转换，但是没有在后面的action中使用对应的结果，在执行时会被跳过。
• 三者都有partition的概念，都有缓存(cache)的操作,还可以进行检查点操作(checkpoint)
• 三者都有许多共同的函数(如map、filter，sorted等等)。
  在对DataFrame和DataSet操作的时候，大多数情况下需要引入隐式转换(ssc.implicits._)

2）区别

• DataFrame：DataFrame是DataSet的特例，也就是说DataSet[Row]的别名；DataFrame = RDD + schema
  1. DataFrame的每一行的固定类型为Row，只有通过解析才能获得各个字段的值
  2. DataFrame与DataSet通常与spark ml同时使用
  3. DataFrame与DataSet均支持sparkSql操作，比如select，groupby等，也可以注册成临时表，进行sql语句操作
  4. DataFrame与DateSet支持一些方便的保存方式，比如csv，可以带上表头，这样每一列的字段名就可以一目了然
• DataSet：DataSet = RDD + case class
  1. DataSet与DataFrame拥有相同的成员函数，区别只是只是每一行的数据类型不同。
  2. DataSet的每一行都是case class，在自定义case class之后可以很方便的获取每一行的信息

五、spark-shell

Spark的shell作为一个强大的交互式数据分析工具，提供了一个简单的方式学习API。它可以使用Scala(在Java虚拟机上运行现有的Java库的一个很好方式)或Python。spark-shell的本质是在后台调用了spark-submit脚本来启动应用程序的，在spark-shell中会创建了一个名为sc的SparkContext对象。

【注】spark-shell只能以client方式启动。

查看帮助

$ spark-shell --help

spark-shell常用选项

--master MASTER_URL 指定模式（spark://host:port, mesos://host:port, yarn,
                              k8s://https://host:port, or local (Default: local[*])）
--executor-memory MEM 指定每个Executor的内存，默认1GB
--total-executor-cores NUM 指定所有Executor所占的核数
--num-executors NUM 指定Executor的个数
--help, -h 显示帮助信息
--version 显示版本号

从上面帮助看，spark有五种运行模式：spark、mesos、yarn、k8s、local。这里主要讲local和yarn模式

1）local

在Spark Shell中，有一个专有的SparkContext已经为您创建好了，变量名叫做sc。自己创建的SparkContext将无法工作。可以用–master参数来设置SparkContext要连接的集群，用–jars来设置需要添加到CLASSPATH的jar包，如果有多个jar包，可以使用逗号分隔符连接它们。例如，在一个拥有2核的环境上运行spark-shell，使用：

#资源存储的位置，默认为本地，以及使用什么调度框架 ，默认使用的是spark内置的资源管理和调度框架Standalone 
# local单机版，只占用一个线程,local[*]占用当前所有线程，local[2]：2个CPU核运行
$ spark-shell --master local[2]
# --master 默认为 local[*] 
#默认使用集群最大的内存大小
--executor-memorty
#默认使用最大核数
--total-executor-cores 
$ spark-shell --master local[*] --executor-memory 1g --total-executor-cores 1

Web UI地址：http://hadoop-node1:4040

随后,就可以使用spark-shell内使用Scala语言完成一定的操作。这里做几个简单的操作，有兴趣的话，可以自行去了解scala

val textFile = sc.textFile("file:///opt/bigdata/hadoop/server/spark-3.2.0-bin-hadoop3.2/README.md")
textFile.count()
textFile.first()

其中，count代表RDD中的总数据条数；first代表RDD中的第一行数据。

2）on Yarn（推荐）

# on yarn，也可以在配置文件中修改这个字段spark.master
$ spark-shell --master yarn 

–master用来设置context将要连接并使用的资源主节点，master的值是standalone模式中spark的集群地址、yarn或mesos集群的URL，或是一个local地址。

六、SparkSQL和Hive的集成（Spark on Hive）

1）创建软链接

$ ln -s /opt/bigdata/hadoop/server/apache-hive-3.1.2-bin/conf/hive-site.xml /opt/bigdata/hadoop/server/spark-3.2.0-bin-hadoop3.2/conf/hive-site.xml

2）复制 hive lib目录 下的mysql连接jar包到spark的jars下

$ cp /opt/bigdata/hadoop/server/apache-hive-3.1.2-bin/lib/mysql-connector-java-5.1.49-bin.jar /opt/bigdata/hadoop/server/spark-3.2.0-bin-hadoop3.2/jars/

3）配置

# 创建spark日志在hdfs存储目录
$ hadoop fs -mkdir -p /tmp/spark
$ cd /opt/bigdata/hadoop/server/spark-3.2.0-bin-hadoop3.2/conf
$ cp spark-defaults.conf.template spark-defaults.conf

在spark-defaults.conf追加如下配置：

# 使用yarn模式
spark.master                     yarn
spark.eventLog.enabled           true
spark.eventLog.dir               hdfs://hadoop-node1:8082/tmp/spark
spark.serializer                 org.apache.spark.serializer.KryoSerializer
spark.driver.memory              512m
spark.executor.extraJavaOptions  -XX:+PrintGCDetails -Dkey=value -Dnumbers="one two three"

4）启动 spark-shell操作Hive（local）

支持多用户得启动metastore服务

$ nohup hive --service metastore &
$ ss -atnlp|grep 9083

在hive-site.xml加入如下配置：

<property>  
  <name>hive.metastore.uris</name>  
  <value>thrift://hadoop-node1:9083</value>  
</property>  

启动spark-sql

# yarn模式，--master yarn可以不带，因为上面在配置文件里已经配置了yarn模式了
$ spark-sql --master yarn
show databases;

从上图就可发现，已经查到我之前创建的库了，说明已经集成ok了。

七、Spark beeline

Spark Thrift Server 是 Spark 社区基于 HiveServer2 实现的一个 Thrift 服务。旨在无缝兼容
HiveServer2。因为 Spark Thrift Server 的接口和协议都和 HiveServer2 完全一致，因此我们部署好Spark Thrift Server后，可以直接使用hive的beeline访问Spark Thrift Server执行相关语句。Spark Thrift Server 的目的也只是取代 HiveServer2，因此它依旧可以和 Hive Metastore进行交互，获取到 hive 的元数据。

1）Spark Thrift Server架构于HiveServer2架构对比

2）Spark Thrift Server和HiveServer2的区别

【总结】Spark Thrift Server说白了就是小小的改动了下HiveServer2，代码量也不多。虽然接口和HiveServer2完全一致，但是它以单个Application在集群运行的方式还是比较奇葩的。可能官方也是为了实现简单而没有再去做更多的优化。

3）配置启动Spark Thrift Server

1、配置hive-site.xml

<!-- hs2端口 -->
<property>
  <name>hive.server2.thrift.port</name>
  <value>11000</value>
</property>

2、启动spark thriftserver服务（不能起hs2，因为配置是一样的，会有冲突）

$ cd /opt/bigdata/hadoop/server/spark-3.2.0-bin-hadoop3.2/sbin
$ ./start-thriftserver.sh
$ ss -tanlp|grep 11000

3、启动beeline操作

# 为了和hive的区别，这里使用绝对路径启动
$ cd /opt/bigdata/hadoop/server/spark-3.2.0-bin-hadoop3.2/bin
# 操作跟hive操作一模一样，只是计算引擎不一样了，换成了spark了
$ ./beeline
!connect jdbc:hive2://hadoop-node1:11000
show databases;

访问HDFS WEB UI：http://hadoop-node1:8088/cluster/apps

八、Spark Streaming

Spark Streaming与其他大数据框架Storm、Flink一样，Spark Streaming是基于Spark Core基础之上用于处理实时计算业务的框架。其实现就是把输入的流数据进行按时间切分，切分的数据块用离线批处理的方式进行并行计算处理。原理如下图：

支持多种数据源获取数据：

Spark处理的是批量的数据（离线数据），Spark Streaming实际上处理并不是像Strom一样来一条处理一条数据，而是将接收到的实时流数据，按照一定时间间隔，对数据进行拆分，交给Spark Engine引擎，最终得到一批批的结果。

由于考虑到本篇文章篇幅太长，所以这里只是稍微提了一下，如果有时间会继续补充Spark Streaming相关的知识点，请耐心等待……

官方文档：https://spark.apache.org/docs/3.2.0/streaming-programming-guide.html