09.大数据技术之Spark

文章目录

• 一、Spark概述
• • 1、概述
  • 2、Spark特点
• 二、Spark角色介绍及运行模式
• • 1、集群角色
  • 2、运行模式
• 三、Spark集群安装
• • 1.Local模式
  • • 1.下载文件
    • 2.解压缩
    • 3、修改配置文件
    • 4.配置环境变量
    • 5.启动服务
    • 6.启动客户端
  • 2.Standalone模式
  • • 1.停止服务
    • 2.修改配置文件spark-env.sh
    • 3.修改配置文件workers
    • 4、将配置好后的spark-3.1.2安装包分发到其他节点
    • 5、分发/etc/profile
    • 5.启动服务
    • 6.测试
    • • 1.spark-shell
      • 2.spark-submit
  • 3.Spark on Yarn
  • • 3.1yarn-client提交任务方式
    • 3.1yarn-cluster提交任务方式
• 四、Spark Core
• • 1、RDD概述
  • 2、 RDD的属性
  • 3、 RDD的弹性
  • 4、RDD特点
  • 5.代码示例
  • • 1.创建Maven工程，添加依赖
• 五、Spark Streaming
• • 1、 Spark streaming简介
  • 2、 Dstream(离散流)
  • 3、 数据源
  • 4.代码示例
  • • 1、linux服务器安装nc服务
    • 2、创建Maven工程，添加依赖
    • 3、代码实现

一、Spark概述 1、概述

Spark是一种快速、通用、可扩展的大数据分析引擎，2009年诞生于加州大学伯克利分校AMPLab，2010年开源，2013年6月成为Apache孵化项目，2014年2月成为Apache顶级项目。项目是用Scala进行编写。
目前，Spark生态系统已经发展成为一个包含多个子项目的集合，其中包含SparkCore、SparkSQL、Spark Streaming、GraphX、MLib、SparkR等子项目，Spark是基于内存计算的大数据并行计算框架。除了扩展了广泛使用的 MapReduce 计算模型，而且高效地支持更多计算模式，包括交互式查询和流处理。Spark 适用于各种各样原先需要多种不同的分布式平台的场景，包括批处理、迭代算法、交互式查询、流处理。通过在一个统一的框架下支持这些不同的计算，Spark 使我们可以简单而低耗地把各种处理流程整合在一起。而这样的组合，在实际的数据分析 过程中是很有意义的。不仅如此，Spark 的这种特性还大大减轻了原先需要对各种平台分 别管理的负担。

Spark Core：实现了 Spark 的基本功能，包含任务调度、内存管理、错误恢复、与存储系统交互等模块。Spark Core 中还包含了对弹性分布式数据集(resilient distributed dataset，简称RDD)的 API 定义。
Spark SQL：是 Spark 用来操作结构化数据的程序包。通过 Spark SQL，我们可以使用 SQL 或者 Apache Hive 版本的 SQL 方言(HQL)来查询数据。Spark SQL 支持多种数据源，比 如 Hive 表、Parquet 以及 JSON 等。
Spark Streaming：是 Spark 提供的对实时数据进行流式计算的组件。提供了用来操作数据流的 API，并且与 Spark Core 中的 RDD API 高度对应。
Spark MLlib：提供常见的机器学习(ML)功能的程序库。包括分类、回归、聚类、协同过滤等，还提供了模型评估、数据 导入等额外的支持功能。
GraphX：提供一个分布式图计算框架，能高效进行图计算。
集群管理器：Spark设计为可以高效地在一个计算节点到数千个计算节点之间伸缩计算。为了实现这样的要求，同时获得最大灵活性，Spark 支持在各种集群管理器(cluster manager)上运行，包括 Hadoop YARN、Apache Mesos，以及 Spark 自带的一个简易调度 器，叫作独立调度器Standalone。
Spark得到了众多大数据公司的支持，这些公司包括Hortonworks、IBM、Intel、Cloudera、MapR、Pivotal、百度、阿里、腾讯、京东、携程、优酷土豆。当前百度的Spark已应用于凤巢、大搜索、直达号、百度大数据等业务；阿里利用GraphX构建了大规模的图计算和图挖掘系统，实现了很多生产系统的推荐算法；腾讯Spark集群达到8000台的规模，是当前已知的世界上最大的Spark集群。

2、Spark特点

1）快：与Hadoop的MapReduce相比，Spark基于内存的运算要快100倍以上，基于硬盘的运算也要快10倍以上。Spark实现了高效的DAG执行引擎，可以通过基于内存来高效处理数据流。计算的中间结果是存在于内存中的。
2）易用：Spark支持Java、Python和Scala的API，还支持超过80种高级算法，使用户可以快速构建不同的应用。而且Spark支持交互式的Python和Scala的shell，可以非常方便地在这些shell中使用Spark集群来验证解决问题的方法。
3）通用：Spark提供了统一的解决方案。Spark可以用于批处理、交互式查询（Spark SQL）、实时流处理（Spark Streaming）、机器学习（Spark MLlib）和图计算（GraphX）。这些不同类型的处理都可以在同一个应用中无缝使用。Spark统一的解决方案非常具有吸引力，毕竟任何公司都想用统一的平台去处理遇到的问题，减少开发和维护的人力成本和部署平台的物力成本。
4）兼容性：Spark可以非常方便地与其他的开源产品进行融合。比如，Spark可以使用Hadoop的YARN和Apache Mesos作为它的资源管理和调度器，并且可以处理所有Hadoop支持的数据，包括HDFS、HBase和Cassandra等。这对于已经部署Hadoop集群的用户特别重要，因为不需要做任何数据迁移就可以使用Spark的强大处理能力。Spark也可以不依赖于第三方的资源管理和调度器，它实现了Standalone作为其内置的资源管理和调度框架，这样进一步降低了Spark的使用门槛，使得所有人都可以非常容易地部署和使用Spark。此外，Spark还提供了在EC2上部署Standalone的Spark集群的工具。

二、Spark角色介绍及运行模式 1、集群角色

从物理部署层面上来看，Spark主要分为两种类型的节点，Master节点和Worker节点：Master节点主要运行集群管理器的中心化部分，所承载的作用是分配Application到Worker节点，维护Worker节点，Driver，Application的状态。Worker节点负责具体的业务运行。
从Spark程序运行的层面来看，Spark主要分为驱动器节点和执行器节点。

2、运行模式

运行环境	模式	模式
Local	本地模式	常用于本地开发测试，如在eclipse，idea中写程序测试等。本地还分为local单线程和local-cluster多线程local[*]
Standalone	集群模式	Spark自带的一个资源调度框架，支持完全分布式。存在的Master单点故障可由ZooKeeper来实现HA
Yarn	集群模式	运行在yarn资源管理器框架之上，由yarn负责资源管理，Spark负责任务调度和计算

1）Local模式: Local模式就是运行在一台计算机上的模式，通常就是用于在本机上练手和测试。它可以通过以下集中方式设置master。
local: 所有计算都运行在一个线程当中，没有任何并行计算，通常我们在本机执行一些测试代码，或者练手，就用这种模式;
local[K]: 指定使用几个线程来运行计算，比如local[4]就是运行4个worker线程。通常我们的cpu有几个core，就指定几个线程，最大化利用cpu的计算能力;
local[*]: 这种模式直接帮你按照cpu最多cores来设置线程数了。
2）Standalone模式: 构建一个由Master+Slave构成的Spark集群，Spark运行在集群中。

3）Yarn模式: Spark客户端直接连接Yarn；不需要额外构建Spark集群。有yarn-client和yarn-cluster两种模式，主要区别在于：Driver程序的运行节点。
yarn-client：Driver程序运行在客户端，适用于交互、调试，希望立即看到app的输出
yarn-cluster：Driver程序运行在由RM（ResourceManager）启动的AP（APPMaster中，这种模式适用于生产环境

三、Spark集群安装 1.Local模式 1.下载文件

下载地址：

http://spark.apache.org/downloads.html

或者

https://archive.apache.org/dist/spark/

选择合适自己的版本下载。我们这里选择spark-3.1.2-bin-hadoop2.7.tgz版本。

2.解压缩

将我们的压缩包上传到hadoop01服务器的/opt/softwares路径下并解压

cd /opt/softwares/

tar -xvzf spark-3.1.2-bin-hadoop2.7.tgz -C ../servers/

3、修改配置文件

修改spark/conf/spark-env.sh.template,执行如下命令：

cd /opt/servers/spark-3.1.2-bin-hadoop2.7/conf

cp spark-env.sh.template spark-env.sh
vim spark-env.sh

新增如下内容：

export JAVA_HOME=/opt/servers/jdk1.8.0_65

4.配置环境变量

修改/etc/profile文件，新增spark环境变量：

export SPARK_HOME=/opt/servers/spark-3.1.2-bin-hadoop2.7
export PATH=:$PATH:$SPARK_HOME/bin:$SPARK_HOME/sbin

修改完成后记得执行 source /etc/profile 使其生效

5.启动服务

cd /opt/servers/spark-3.1.2-bin-hadoop2.7
sbin/start-all.sh

6.启动客户端

spark-shell 

简单测试

读取spark安装目录下的readme.md文件，并统计词条数量和显示第一行字符。

scala> val textFile = sc.textFile("file:///opt/servers/spark-3.1.2-bin-hadoop2.7/README.md")   //读取readme.md文件
textFile: org.apache.spark.rdd.RDD[String] = /opt/servers/spark-3.1.2-bin-hadoop2.7/README.md MapPartitionsRDD[1] at textFile at :24

scala> textFile.count()   //词条统计
res0: Long = 108                                                                

scala> textFile.first()   //打印第一行字符
res1: String = # Apache Spark

2.Standalone模式

Standalone模式可以在Local模式下继续操作。

1.停止服务

cd /opt/servers/spark-3.1.2-bin-hadoop2.7
sbin/stop-all.sh

2.修改配置文件spark-env.sh

修改spark-env.sh文件，执行以下操作

cp spark-env.sh.template spark-env.sh
vim spark-env.sh

添加以下内容：

export JAVA_HOME=/opt/servers/jdk1.8.0_65
SPARK_MASTER_HOST=hadoop01
SPARK_MASTER_PORT=7077

3.修改配置文件workers

mv workers.template workers
vim workers

添加以下内容：

hadoop01
hadoop02
hadoop03

4、将配置好后的spark-3.1.2安装包分发到其他节点

cd /opt/servers/
scp -r spark-3.1.2-bin-hadoop2.7 hadoop02:$PWD
scp -r spark-3.1.2-bin-hadoop2.7 hadoop03:$PWD

5、分发/etc/profile

分发/etc/profile文件

scp /etc/profile hadoop02:/etc
scp /etc/profile hadoop03:/etc

完成后记得执行 source /etc/profile 使其生效

5.启动服务

cd /opt/servers/spark-3.1.2-bin-hadoop2.7
sbin/start-all.sh

6.测试 1.spark-shell

spark-shell 
--master spark://hadoop01:7077 
--executor-memory 1g 
--total-executor-cores 2

读取spark安装目录下的readme.md文件，并统计词条数量和显示第一行字符。

scala> val textFile = sc.textFile("hdfs://hadoop01:8020/test/input/README.md")   //读取readme.md文件
textFile: org.apache.spark.rdd.RDD[String] = /opt/servers/spark-3.1.2-bin-hadoop2.7/README.md MapPartitionsRDD[1] at textFile at :24

scala> textFile.count()   //词条统计
res0: Long = 108                                                                

scala> textFile.first()   //打印第一行字符
res1: String = # Apache Spark

2.spark-submit

该算法是利用蒙特·卡罗算法求PI

cd /opt/servers/spark-3.1.2-bin-hadoop2.7


bin/spark-submit 
--class org.apache.spark.examples.SparkPi 
--master spark://hadoop01:7077 
./examples/jars/spark-examples_2.12-3.1.2.jar 
10

3.Spark on Yarn

yarn模式两种提交任务方式：

Spark可以和Yarn整合，将Application提交到Yarn上运行，Yarn有两种提交任务的方式。

3.1yarn-client提交任务方式

bin/spark-submit --class org.apache.spark.examples.SparkPi --master yarn --deploy-mode client ./examples/jars/spark-examples_2.12-3.1.2.jar 100

3.1yarn-cluster提交任务方式

bin/spark-submit --class org.apache.spark.examples.SparkPi --master yarn --deploy-mode cluster ./examples/jars/spark-examples_2.12-3.1.2.jar 100

注：如果在yarn集群中运行报错，可能是因为yarn内存检测的原因导致，需要增加如下配置

vi yarn-site.xml

        
        
                yarn.nodemanager.pmem-check-enabled
                false
        
        
        
                yarn.nodemanager.vmem-check-enabled
                false
        

四、Spark Core 1、RDD概述

什么是RDD?
RDD（Resilient Distributed Dataset）叫做分布式数据集，是Spark中最基本的数据抽象，它代表一个不可变、可分区、弹性、里面的元素可并行计算的集合。在 Spark 中，对数据的所有操作不外乎创建 RDD、转化已有RDD 以及调用 RDD 操作进行求值。每个 RDD 都被分为多个分区，这些分区运行在集群中的不同节点上。RDD 可以包含 Python、Java、Scala 中任意类型的对象， 甚至可以包含用户自定义的对象。RDD具有数据流模型的特点：自动容错、位置感知性调度和可伸缩性。RDD允许用户在执行多个查询时显式地将工作集缓存在内存中，后续的查询能够重用工作集，这极大地提升了查询速度。
RDD支持两种操作:转化操作和行动操作。RDD 的转化操作是返回一个新的 RDD的操作，比如 map()和 filter()，而行动操作则是向驱动器程序返回结果或把结果写入外部系统的操作。比如 count() 和 first()。
Spark采用惰性计算模式，RDD只有第一次在一个行动操作中用到时，才会真正计算。Spark可以优化整个计算过程。默认情况下，Spark 的 RDD 会在你每次对它们进行行动操作时重新计算。如果想在多个行动操作中重用同一个 RDD，可以使用 RDD.persist() 让 Spark 把这个 RDD 缓存下来。

2、 RDD的属性

1. 一组分片（Partition），即数据集的基本组成单位。对于RDD来说，每个分片都会被一个计算任务处理，并决定并行计算的粒度。用户可以在创建RDD时指定RDD的分片个数，如果没有指定，那么就会采用默认值。默认值就是程序所分配到的CPU Core的数目。
2. 一个计算每个分区的函数。Spark中RDD的计算是以分片为单位的，每个RDD都会实现compute函数以达到这个目的。compute函数会对迭代器进行复合，不需要保存每次计算的结果。
3. RDD之间的依赖关系。RDD的每次转换都会生成一个新的RDD，所以RDD之间就会形成类似于流水线一样的前后依赖关系。在部分分区数据丢失时，Spark可以通过这个依赖关系重新计算丢失的分区数据，而不是对RDD的所有分区进行重新计算。
4. 一个Partitioner，即RDD的分片函数。当前Spark中实现了两种类型的分片函数，一个是基于哈希的HashPartitioner，另外一个是基于范围的RangePartitioner。只有对于key-value的RDD，才会有Partitioner，非key-value的RDD的Parititioner的值是None。Partitioner函数不但决定了RDD本身的分片数量，也决定了parent RDD Shuffle输出时的分片数量。
5. 一个列表，存储存取每个Partition的对应数据的优先位置（preferred location）。对于一个HDFS文件来说，这个列表保存的就是每个Partition所在的块的位置。按照“移动数据不如移动计算”的理念，Spark在进行任务调度的时候，会尽可能地将计算任务分配到其所要处理数据块的存储位置。

RDD是一个应用层面的逻辑概念。一个RDD多个分片。RDD就是一个元数据记录集，记录了RDD内存所有的关系数据。

3、 RDD的弹性

1. 自动进行内存和磁盘数据存储的切换
   Spark优先把数据放到内存中，如果内存放不下，就会放到磁盘里面，程序进行自动的存储切换

2. 基于血统的高效容错机制
   在RDD进行转换和动作的时候，会形成RDD的Lineage依赖链，当某一个RDD失效的时候，可以通过重新计算上游的RDD来重新生成丢失的RDD数据，如果只是某个分区的数据丢失，只计算这个分区的数据。

3. Task如果失败会自动进行特定次数的重试
   RDD的计算任务如果运行失败，会自动进行任务的重新计算，默认次数是4次。

4. Stage如果失败会自动进行特定次数的重试
   如果Job的某个Stage阶段计算失败，框架也会自动进行任务的重新计算，默认次数也是4次。
   Spark中的Stage其实就是一组并行的任务，任务是一个个的task

5. Checkpoint和Persist可主动或被动触发
   RDD可以通过Persist持久化将RDD缓存到内存或者磁盘，当再次用到该RDD时直接读取就行。也可以将RDD进行检查点，检查点会将数据存储在HDFS中，该RDD的所有父RDD依赖都会被移除。

6. 数据调度弹性
   Spark把这个JOB执行模型抽象为通用的有向无环图DAG，可以将多Stage的任务串联或并行执行，调度引擎自动处理Stage的失败以及Task的失败。

4、RDD特点

RDD表示只读的分区的数据集，对RDD进行改动，只能通过RDD的转换操作，由一个RDD得到一个新的RDD，新的RDD包含了从其他RDD衍生所必需的信息。RDDs之间存在依赖，RDD的执行是按照血缘关系延时计算的。如果血缘关系较长，可以通过持久化RDD来切断血缘关系。
（1）分区
RDD逻辑上是分区的，每个分区的数据是抽象存在的，计算的时候会通过一个compute函数得到每个分区的数据。如果RDD是通过已有的文件系统构建，则compute函数是读取指定文件系统中的数据，如果RDD是通过其他RDD转换而来，则compute函数是执行转换逻辑将其他RDD的数据进行转换。

（2）只读
如下图所示，RDD是只读的，要想改变RDD中的数据，只能在现有的RDD基础上创建新的RDD。

由一个RDD转换到另一个RDD，可以通过丰富的操作算子实现，不再像MapReduce那样只能写map和reduce了，如下图所示。

RDD的操作算子包括两类，一类叫做transformations，它是用来将RDD进行转化，构建RDD的血缘关系；另一类叫做actions，它是用来触发RDD的计算，得到RDD的相关计算结果或者将RDD保存的文件系统中。下图是RDD所支持的操作算子列表。
（3）依赖
RDDs通过操作算子进行转换，转换得到的新RDD包含了从其他RDDs衍生所必需的信息，RDDs之间维护着这种血缘关系，也称之为依赖。如下图所示，依赖包括两种，一种是窄依赖，RDDs之间分区是一一对应的，另一种是宽依赖，下游RDD的每个分区与上游RDD(也称之为父RDD)的每个分区都有关，是多对多的关系，宽依赖会引起shuffle。

通过RDDs之间的这种依赖关系，一个任务流可以描述为DAG(有向无环图)，如下图所示，在实际执行过程中宽依赖对应于Shuffle(图中的reduceByKey和join)，窄依赖中的所有转换操作可以通过类似于管道的方式一气呵成执行(图中map和union可以一起执行)。

5.代码示例

需求：统计统计词的出现次数

1.创建Maven工程，添加依赖

    org.apache.spark
    spark-core_2.12
    3.1.2

代码实现

 public static void main(String[] args) {

        SparkConf sparkConf = new SparkConf().setAppName("JavaWordCount").setMaster("local[*]");

        JavaSparkContext sc = new JavaSparkContext(sparkConf);

        JavaRDD lines = sc.textFile("hdfs://hadoop01:8020/test/input/word.txt");

        JavaRDD words = lines.flatMap(new FlatMapFunction() {
            @Override
            public Iterator call(String line) throws Exception {
                return Arrays.asList(line.split(" ")).iterator();
            }
        });

        JavaPairRDD word = words.mapToPair(new PairFunction() {

            @Override
            public Tuple2 call(String word) throws Exception {
                return new Tuple2<>(word, 1);
            }
        });

        JavaPairRDD wordcount = word.reduceByKey(new Function2() {

            @Override
            public Integer call(Integer i1, Integer i2) throws Exception {
                return i1 + i2;
            }
        });

        wordcount.foreach(new VoidFunction>() {
            public void call(Tuple2 o) throws Exception {
                System.out.println(o._1 + " : " + o._2);
            }
        });
    }

五、Spark Streaming 1、 Spark streaming简介

Spark Streaming是Spark API的核心扩展，支持实时数据流的可扩展、高吞吐量和容错流处理。数据可以从Kafka、Kinesis或TCP套接字等多种来源中获取，并且可以使用复杂的算法进行处理，这些算法用高级函数表示，如map、reduce、join和window。最后，处理过的数据可以推送到文件系统、数据库和实时仪表板。事实上，您可以在数据流上应用Spark的机器学习和图形处理算法。

在内部，它的工作方式如下。Spark Streaming接收实时输入的数据流，并对数据进行分批处理，由Spark引擎进行处理，生成最终的批量结果流。

Spark Streaming提供了一种高级抽象，称为离散流或DStream，它表示连续的数据流。Dstream可以通过来自Kafka和Kinesis等源的输入数据流创建，也可以通过在其他Dstream上应用高级操作来创建。在内部，DStream表示为rdd序列。

2、 Dstream(离散流)

Dstream是Spark Streaming的数据抽象，同DataFrame，其实底层依旧是RDD。

Discretized Stream或DStream是Spark Streaming提供的基本抽象。它表示一个连续的数据流，要么是从源接收的输入数据流，要么是通过转换输入流生成的处理数据流。在内部，DStream由一系列连续的rdd表示，这是Spark对不可变的分布式数据集的抽象。DStream中的每个RDD都包含一定时间间隔的数据，如下图所示：

在DStream上应用的任何操作都转换为在底层rdd上的操作。

这些底层RDD转换是由Spark引擎计算的。DStream操作隐藏了大部分细节，并为开发人员提供了更高级的API。

3、 数据源

Spark Streaming提供了两类内置流源:

基本源:在StreamingContext API中直接可用的源。例如文件系统和套接字连接。

高级资源:像Kafka, Kinesis等资源可以通过额外的实用程序类获得。这些需要根据链接部分中讨论的额外依赖项进行链接。

注意,如果希望在流应用程序中并行接收多个数据流，可以创建多个输入Dstream。这将创建多个接收器，这些接收器将同时接收多个数据流。但是请注意，Spark worker/executor是一个长期运行的任务，因此它占用分配给Spark Streaming应用程序的一个核心。因此，Spark Streaming应用程序需要分配足够的内核(或者线程，如果在本地运行的话)来处理接收到的数据，以及运行接收端，记住这一点很重要。

4.代码示例

需求：统计统计词的出现次数

1、linux服务器安装nc服务

yum -y install nc

#使用nc命令发送数据
nc -lk 9999

2、创建Maven工程，添加依赖

    org.apache.spark
    spark-streaming_2.12
    3.1.2

3、代码实现

public static void main(String[] args) throws Exception{

        // Create a local StreamingContext with two working thread and batch interval of 1 second
        SparkConf conf = new SparkConf().setMaster("local[2]").setAppName("NetworkWordCount");
        JavaStreamingContext jssc = new JavaStreamingContext(conf, Durations.seconds(5));

        // Create a DStream that will connect to hostname:port, like localhost:9999
        JavaReceiverInputDStream lines = jssc.socketTextStream("hadoop01", 9999);

        lines.flatMap(x -> Arrays.asList(x.split(" ")).iterator());
        // Split each line into words
        JavaDStream words = lines.flatMap(x -> Arrays.asList(x.split(" ")).iterator());
        // Count each word in each batch
        JavaPairDStream pairs = words.mapToPair(s -> new Tuple2<>(s, 1));
        JavaPairDStream wordCounts = pairs.reduceByKey((i1, i2) -> i1 + i2);
        // Print the first ten elements of each RDD generated in this DStream to the console
        wordCounts.print();
        jssc.start();              // Start the computation
        jssc.awaitTermination();   // Wait for the computation to terminate
    }