首页 > 其他分享 >Day11 —— 大数据技术之Spark

Day11 —— 大数据技术之Spark

时间：2024-06-24 13:59:05浏览次数：27  
标签：算子 Hive RDD Day11 SQL Spark 数据 spark

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提交方式

当提交方式为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 more
   contributor license agreements.  See the NOTICE file distributed with
   this 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 with
   the License.  You may obtain a copy of the License at

       http://www.apache.org/licenses/LICENSE-2.0

   Unless required by applicable law or agreed to in writing, software
   distributed 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 and
   limitations 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()
      // 创建表students
      spark.sql("create table if not exists students(id int,name string,age int)row format delimited fields terminated by '\t'")
      // 导入数据到表students
      spark.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生态系统集成紧密。是处理大规模数据的强大工具。

标签：算子,Hive,RDD,Day11,SQL,Spark,数据,spark
From： https://blog.csdn.net/weixin_62804382/article/details/139814251

相关文章

  • 【Python爬虫实战项目】Python爬取Top100电影榜单数据并保存csv文件（附源码）
    前言今天给大家介绍的是Python爬取Top100电影榜单数据保存csv文件，在这里给需要的小伙伴们帮助，并且给出一点小心得。开发工具Python版本：3.6相关模块：requests模块time模块parsel模块csv模块环境搭建安装Python并添加到环境变量，pip安装需要的相关模块即可。文中......
  • [数据集][目标检测]驾驶中眼睛疲劳检测数据集VOC+YOLO格式4030张2类别
    数据集格式：PascalVOC格式+YOLO格式(不包含分割路径的txt文件，仅仅包含jpg图片以及对应的VOC格式xml文件和yolo格式txt文件)图片数量(jpg文件个数)：4030标注数量(xml文件个数)：4030标注数量(txt文件个数)：4030标注类别数：2标注类别名称:["Attentiveeye","Drowsyeye"]每个类......
  • 中国车牌检测数据集VOC+YOLO格式2001张1类别
    数据集格式：PascalVOC格式+YOLO格式(不包含分割路径的txt文件，仅仅包含jpg图片以及对应的VOC格式xml文件和yolo格式txt文件)图片数量(jpg文件个数)：2001标注数量(xml文件个数)：2001标注数量(txt文件个数)：2001标注类别数：1标注类别名称:["plate"]每个类别标注的框数：plate框......
  • [本科项目实训] P-Tuning v2 训练数据生成
    思路由于我们项目的任务并不属于通用的自然语言处理任务，所以显然没有公开数据集，因而手工生成与模型生成相结合的方式或许是一个不错的选择。在不同的使用方式和任务情景下，我们曾经尝试了不同的数据集生成方式：#v1importjsonimportrandomtrain_f=True#train_f=False......
  • Kubernetes面试整理-解释Etcd在Kubernetes中的作用，包括如何管理配置数据和状态信息
    etcd是一个分布式的键值存储系统，在Kubernetes中起着至关重要的作用。它主要用于存储集群的所有配置数据和状态信息，确保这些数据的一致性和高可用性。具体来说，etcd在Kubernetes中的作用如下：etcd的作用● 配置存储：etcd存储Kubernetes集群的所有配置信息，包括节点......
  • 【数据结构与算法】拓扑排序,关键活动,关键路径 详解
    拓扑排序算法booltopologicalSort(){ stack<int>stk; intid[N]; intcnt=0; for(inti=1;i<=n;i++){ if(!inDeg[i]){ stk.push(i); } id[i]=inDeg[i]; } while(stk.size()){ intt=stk.top(); stk.pop(); cout<<t<......
  • 【数据结构与算法】最小生成树,Prim算法,Kruskal算法 详解
    最小生成树的实际应用背景。最节省经费的前提下，在n个城市之间建立通信联络网。Kruskal算法（基于并查集）voidinit(){for(inti=1;i<=n;i++){pre[i]=i;}}llroot(lla){lli=a;while(pre[i]!=i){i=pre[i];......
  • 【JS逆向百例】某点数据逆向分析，多方法详解
    前言最近收到粉丝的私信，其在逆向某个站点时遇到了些问题，在查阅资料未果后，来询问K哥，K哥一向会尽力满足粉丝的需求。网上大多数分析该站点的教程已经不再适用，本文K哥将提供3种解决方案，对于webpack不太熟练的小伙伴来说，这是一个很好的练手案例：逆向目标目标：某点数据，排行榜......
  • MySQL数据库管理
    一：使用MySQL数据库1：查看数据库结构[root@localhost~]#mysql-uroot-ppwd123mysql>showdatabases;mysql>usemysqlmysql>showtables;mysql>describeuser;mysql>select*fromuser\G;备注：information_schema数据库：保存着关于MySQL服务器所维护的所......
  • SSM-学情分析系统-56772（免费领源码+开发文档）可做计算机毕业设计JAVA、PHP、爬虫、APP
    学情分析系统摘 要随着科学技术的飞速发展，各行各业都在努力与现代先进技术接轨，通过科技手段提高自身的优势；对于学情分析系统当然也不能排除在外，随着网络技术的不断成熟，带动了学情分析系统，它彻底改变了过去传统的管理方式，不仅使服务管理难度变低了，还提升了管理的灵活性。这......

赞助商

阅读排行

网站主页关于我们联系我们网站地图

本网站内容转载自其他媒体，侵权联系[admin##ips99.com]。

Copyright © 2020-2023 IPS99 版权所有 IPS99