Day11 —— 大数据技术之Spark

Spark的概述

什么是Spark？

Apache Spark是一个快速通用的集群计算系统，是一种与Hadoop相似的开源集群计算环境，但是Spark在一些工作负载方面表现得更加优越。它提供了Java、Scala、Python和R的高级API，以及一个支持通用的执行图计算的优化引擎。它还支持高级工具，包括使用SQL进行结构化数据处理的Spark SQL、用于机器学习的MLlib、用于图处理的GraphX，以及用于实时流处理的Spark Streaming。

Spark的主要特点

快速性：Spark使用内存计算技术，将数据缓存在内存中，因此比传统的基于磁盘的计算框架快许多倍，适用于需要快速处理大规模数据的场景。

通用性：Spark提供了用于批处理、交互式查询、流处理和机器学习等多种计算模型，使得用户可以在同一个平台上处理各种类型的工作负载。

容错性：Spark通过弹性分布式数据集（RDD）实现容错性，能够在节点发生故障时自动恢复，保证计算的可靠性。

易用性：Spark提供了丰富的API（如Scala、Java、Python和R），并且有丰富的第三方库和工具支持，使得开发者能够快速上手并开发复杂的数据处理应用。

可扩展性：Spark可以在多个节点上并行处理数据，支持水平扩展，能够处理PB级别的数据规模。

流处理能力：Spark Streaming模块支持实时数据处理，能够处理实时数据流并提供低延迟的计算结果。

机器学习支持：Spark提供了MLlib机器学习库，支持常见的机器学习算法，使得在Spark上进行大规模机器学习任务变得更加容易。

Spark的主要组件

Spark是由多个组件构成的软件栈，Spark 的核心（Spark Core）是一个对由很多计算任务组成的、运行在多个工作机器或者一个计算集群上的应用进行调度、分发以及监控的计算引擎。

Spark安装

下载解压缩spark-3.3.3-bin-hadoop3.tgz，重命名Spark安装目录为spark，

在配置文件/etc/profile中添加：

export SPARK_HOME=/export/servers/spark
export PATH=$ PATH:$SPARK_HOME/bin

执行/etc/profile脚本，使配置生效

source /etc/profile

Spark三种运行模式

1. Local Mode（本地模式）：

   在本地模式下，Spark将在单个计算机上运行，适用于开发、测试和小规模数据处理。在本地模式下，Spark将使用单个线程处理数据，不涉及集群。

2. Standalone Mode（独立模式）：

   在独立模式下，Spark使用自带的资源管理器，可以在一个独立的Spark集群上运行。这种模式下，Spark独立管理资源，可以运行在独立模式的集群上，而不依赖于其他资源管理器。

3. Spark On YARN模式：

   在Apache Hadoop的资源管理框架YARN上以分布式方式在集群中运行 Apache Spark

Spark Standalone架构

Spark Standalone的两种提交方式

​ Spark Standalone模式为经典的Master/Slave架构，资源调度是Spark自己实现的。在Standalone模式中，根据应用程序提交的方式不同，Driver（主控进程）在集群中的位置也有所不同。应用程序的提交方式主要有两种：client和cluster，默认是client。可以在向Spark集群提交应用程序时使用–deploy-mode参数指定提交方式。

• client提交方式

• cluster提交方式

Spark On YARN架构

Spark On YARN模式遵循YARN的官方规范，YARN只负责资源的管理和调度，运行哪种应用程序由用户自己决定，因此可能在YARN上同时运行MapReduce程序和Spark程序，YARN对每一个程序很好地实现了资源的隔离。这使得Spark与MapReduce可以运行于同一个集群中，共享集群存储资源与计算资源。

​Spark On YARN模式分为client和cluster两种提交方式。

• client提交方式
  
• cluster提交方式
  

RDD算子

RDD被创建后是只读的，不允许修改。Spark提供了丰富的用于操作RDD的方法，这些方法被称为算子。一个创建完成的RDD只支持两种算子：转化（Transformation）算子和行动（Action）算子。

转化算子

转化算子负责对RDD中的数据进行计算并转化为新的RDD。Spark中的所有转化算子都是惰性的，因为它们不会立即计算结果，而只是记住对某个RDD的具体操作过程，直到遇到行动算子才会与其一起执行。

• map()算子
  map()是一种转化算子，它接收一个函数作为参数，并把这个函数应用于RDD的每个元素，最后将函数的返回结果作为结果RDD中对应元素的值。

• filter()算子
  filter()算子通过函数对源RDD的每个元素进行过滤，并返回一个新的RDD。

• flatMap()算子
  与map()算子类似，但是每个传入函数的RDD元素会返回0到多个元素，最终会将返回的所有元素合并到一个RDD。

• reduceByKey()算子
  reduceByKey()算子的作用对象是元素为(key,value)形式（Scala元组）的RDD，使用该算子可以将key相同的元素聚集到一起，最终把所有key相同的元素合并成一个元素。该元素的key不变，value可以聚合成一个列表或者进行求和等操作。最终返回的RDD的元素类型和原有类型保持一致。

• groupByKey()算子
  groupByKey()算子的作用对象是元素为(key,value)形式（Scala元组）的RDD，使用该算子可以将key相同的元素聚集到一起，最终把所有key相同的元素合并成为一个元素。该元素的key不变，value则聚集到一个集合中。

• union()算子
  union()算子将两个RDD合并为一个新的RDD，主要用于对不同的数据来源进行合并，两个RDD中的数据类型要保持一致。

• sortBy()算子
  sortBy()算子将RDD中的元素按照某个规则进行排序。该算子的第一个参数为排序函数，第二个参数是一个布尔值，指定升序（默认）或降序。若需要降序排列，则需将第二个参数置为false。

• sortByKey()算子
  sortByKey()算子将(key,value)形式的RDD按照key进行排序。默认升序，若需降序排列，则可以传入参数false

• join()算子
  join()算子将两个(key,value)形式的RDD根据key进行连接操作，相当于数据库的内连接（Inner Join），只返回两个RDD都匹配的内容。例如

• distinct()算子
  distinct()算子对RDD中的数据进行去重操作，返回一个新的RDD。

行动算子

Spark中的转化算子并不会马上进行运算，而是在遇到行动算子时才会执行相应的语句，触发Spark的任务调度。

• reduce()算子
  将数字1~100所组成的集合转为RDD，然后对该RDD使用reduce()算子进行计算，统计RDD中所有元素值的总和。
• count()算子
  统计RDD集合中元素的数量。
• countByKey()算子
  List集合中存储的是键值对形式的元组，使用该List集合创建一个RDD，然后对其使用countByKey()算子进行计算。
• take(n)算子
  返回集合中前n个元素组成的数组。

Spark RDD

RDD的创建

RDD中的数据来源可以是程序中的对象集合，也可以是外部存储系统中的数据集，例如共享文件系统、HDFS、HBase或任何提供Hadoop InputFormat的数据源。

从对象集合创建RDD

Spark可以通过parallelize()或makeRDD()方法将一个对象集合转化为RDD。

例如，将一个List集合转化为RDD，代码如下：

val rdd=sc.parallelize(List(1,2,3,4,5,6))
或者
val rdd=sc.makeRDD(List(1,2,3,4,5,6))

从返回信息可以看出，上述创建的RDD中存储的是Int类型的数据。实际上，RDD也是一个集合，与常用的List集合不同的是，RDD集合的数据分布于多台机器上。

从外部存储创建RDD

Spark的textFile()方法可以读取本地文件系统或外部其他系统中的数据，并创建RDD。不同的是，数据的来源路径不同。
读取本地系统文件

将读取的本地文件内容转为一个RDD
val rdd = sc.textFile(“file:///root/data/words.txt”)
使用collect()方法查看RDD中的内容
rdd.collect() # 或者使用rdd.collect

读取HDFS系统文件

将读取的HDFS系统文件内容转为一个RDD
val rdd = sc.textFile(“hdfs://192.168.121.131:9000/words.txt”)
使用collect()方法查看RDD中的内容
rdd.collect() # 或者使用rdd.collect

Spark SQL

Spark SQL是一个用于结构化数据处理的Spark组件。所谓结构化数据，是指具有Schema信息的数据，例如JSON、Parquet、Avro、CSV格式的数据。与基础的Spark RDD API不同，Spark SQL提供了对结构化数据的查询和计算接口。

Spark SQL的基本使用

Spark Shell启动时除了默认创建一个名为sc的SparkContext的实例外，还创建了一个名为spark的SparkSession实例，该spark变量可以在Spark Shell中直接使用。

​SparkSession只是在SparkContext基础上的封装，应用程序的入口仍然是SparkContext。SparkSession允许用户通过它调用DataFrame和Dataset相关API来编写Spark程序，支持从不同的数据源加载数据，并把数据转换成DataFrame，然后使用SQL语句来操作DataFrame数据。

Spark SQL基本使用案例

进入spark-shell环境

加载数据为Dataset

val d1 = spark.read.textFile(“hdfs://192.168.121.131:9000/input/person.txt”)
d1.show() # 查看d1中的数据内容

给Dataset添加元数据信息

case class Person(id:Int,name:String,age:Int)

调用Dataset的map()算子将每一个元素拆分并存入Person类中

val personDataset = d1.map(line=>{
val fields = line.split(“,”)
val id = fields(0).toInt
val name = fields(1)
val age = fields(2).toInt
Person(id,name,age)
})
personDataset.show() # 查看personDataset中的数据内容

将Dataset转为DataFrame

val pdf = personDataset.toDF()

执行SQL查询

pdf.createTempView(“v_person”)
val result = spark.sql(“select * from v_person order by age desc”)
result.show()

Spark SQL函数

• 内置函数
  Spark SQL内置了大量的函数，位于API org.apache.spark.sql.functions
  中。其中大部分函数与Hive中的相同。使用内置函数有两种方式：一种是通过编程的方式使用；另一种是在SQL语句中使用。

• 自定义函数
  当Spark SQL提供的内置函数不能满足查询需求时，用户可以根据需求编写自定义函数（User Defined Functions, UDF），然后在Spark SQL中调用。

• 窗口（开窗）函数
  开窗函数是为了既显示聚合前的数据，又显示聚合后的数据，即在每一行的最后一列添加聚合函数的结果。开窗口函数有以下功能：

  1. 同时具有分组和排序的功能
  2. 不减少原表的行数
  3. 开窗函数语法：

Spark SQL整合Hive

Hive是一个基于Hadoop的数据仓库架构，使用SQL语句读、写和管理大型分布式数据集。Hive可以将SQL语句转化为MapReduce（或Apache Spark、Apache Tez）任务执行，大大降低了Hadoop的使用门槛，减少了开发MapReduce程序的时间成本。可以将Hive理解为一个客户端工具，它提供了一种类SQL查询语言，称为HiveQL。这使得Hive十分适合数据仓库的统计分析，能够轻松使用HiveQL开启数据仓库任务，如提取／转换／加载（ETL）、分析报告和数据分析。Hive不仅可以分析HDFS文件系统中的数据，也可以分析其他存储系统（例如HBase）中的数据。

Spark SQL与Hive整合后，可以在Spark SQL中使用HiveQL轻松操作数据仓库。与Hive不同的是，Hive的执行引擎为MapReduce，而Spark SQL的执行引擎为Spark RDD。

Spark SQL整合Hive的步骤

将$ HIVE_HOME/conf中的hive-site.xml文件复制到$ SPARK_HOME/conf
添加如下属性配置：

<?xml version="1.0" encoding="UTF-8" standalone="no"?>
<?xml-stylesheet type="text/xsl" href="configuration.xsl"?><!--Licensed to the Apache Software Foundation (ASF) under one or morecontributor license agreements.  See the NOTICE file distributed withthis work for additional information regarding copyright ownership.The ASF licenses this file to You under the Apache License, Version 2.0(the "License"); you may not use this file except in compliance withthe License.  You may obtain a copy of the License athttp://www.apache.org/licenses/LICENSE-2.0Unless required by applicable law or agreed to in writing, softwaredistributed under the License is distributed on an "AS IS" BASIS,WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.See the License for the specific language governing permissions andlimitations under the License.
-->
<configuration><!-- 数据库 start --><property><name>javax.jdo.option.ConnectionURL</name><value>jdbc:mysql://localhost:3306/spark_hive_meta?createDatabaseIfNotExist=true&amp;useSSL=false</value><description>mysql连接</description></property><property><name>javax.jdo.option.ConnectionDriverName</name><value>com.mysql.jdbc.Driver</value><description>mysql驱动</description></property><property><name>javax.jdo.option.ConnectionUserName</name><value>root</value><description>数据库使用用户名</description></property><property><name>javax.jdo.option.ConnectionPassword</name><value>123456</value><description>数据库密码</description></property><!-- 数据库 end --><property> <name>hive.metastore.warehouse.dir</name><value>/hive/warehouse</value><description>hive使用的HDFS目录</description></property><property> <name>hive.cli.print.current.db</name><value>true</value></property><property><name>hive.support.concurrency</name><value>true</value><description>开启Hive的并发模式</description></property><property><name>hive.txn.manager</name><value>org.apache.hadoop.hive.ql.lockmgr.DbTxnManager</value><description>用于并发控制的锁管理器类</description></property><property><name>hive.server2.thrift.bind.host</name><value>my2308-host</value><description>hive开启的thriftServer地址</description></property><property><name>hive.server2.thrift.port</name><value>10000</value><description>hive开启的thriftServer端口</description></property><property><name>hive.server2.enable.doAs</name><value>true</value></property><property><name>hive.metastore.schema.verification</name><value>false</value></property><property><name>datanucleus.schema.autoCreateAll</name><value>true</value></property>
</configuration>

在Spark配置文件spark-env.sh中指定Hadoop及其配置文件的主目录（根据自己的Hadoop安装目录修改）。

export HADOOP_HOME=/export/servers/hadoop-3.2.0
export HADOOP_CONF_DIR=$HADOOP_HOME/etc/hadoop

将MySQL驱动JAR包复制到$SPARK_HOME/jars目录中（根据自己的目录复制）。

cp /export/servers/mysql-connector-java-5.1.40/mysql-connector-java-5.1.40-bin.jar /export/servers/spark/jars

Spark SQL操作Hive的三种方式

方式一：Spark SQL终端操作（以Spark Standalone模式为例）

spark-sql --master spark://192.168.121.131:7077

进入Spark SQL终端后，以HiveQL的方式操作

方式二：Spark Shell操作

spark-shell --master spark://192.168.121.131:7077

进入spark shell终端后，通过spark.sql(“HiveQL语句”)的方式操作

方式三：提交Spark SQL应用程序
​
在IDEA中编写Spark SQL操作Hive的应用程序，然后将编写好的应用程序打包为JAR，提交到Spark集群中运行，即对Hive进行数据的读写与分析。

package spark.demo.sql
import org.apache.spark.sql.SparkSession
object SparkSQLHiveDemo {def main(args: Array[String]): Unit = {// 创建SparkSession对象val spark = SparkSession.builder().appName("Spark Hive Demo").enableHiveSupport().getOrCreate()// 创建表studentsspark.sql("create table if not exists students(id int,name string,age int)row format delimited fields terminated by '\t'")// 导入数据到表studentsspark.sql("load data local inpath '/root/data/students.txt' into table students")// 查询表students的数据spark.sql("select * from students").show()}
}

打包程序，然后提交到Spark集群

spark-submit --class spark.demo.sql.SparkSQLHiveDemo SparkDemo-1.0-SNAPSHOT.jar

Spark总结

Apache Spark是一款快速、通用、可扩展的大数据处理框架。其采用内存计算技术，支持批处理、交互式查询、流处理和机器学习等多种数据处理方式。具有高容错性，能够保证作业的正确执行。Spark提供丰富的API和高层次抽象，使用户能够更轻松地编写和调试程序。可在单机或分布式集群上运行，灵活部署，与Hadoop生态系统集成紧密。是处理大规模数据的强大工具。