hadoop离线分析（简单版）spark

目录

概述

spark运行流程图

Spark运行特点

spark组件、功能介绍

Spark常用术语

Spark运行模式

RDD运行流程

Spark安装配置

概述

批处理为主&＃xff0c;流处理为辅--微批

Apache Spark 是专为大规模数据处理而设计的快速通用的计算引擎。Spark是UC Berkeley AMP lab (加州大学伯克利分校的AMP实验室)所开源的类Hadoop MapReduce的通用并行框架&＃xff0c;Spark&＃xff0c;拥有Hadoop MapReduce所具有的优点&＃xff1b;但不同于MapReduce的是——Job中间输出结果可以保存在内存中&＃xff0c;从而不再需要读写HDFS&＃xff0c;因此Spark能更好地适用于数据挖掘与机器学习等需要迭代的MapReduce的算法。

Spark 是一种与Hadoop相似的开源集群计算环境&＃xff0c;但是两者之间还存在一些不同之处&＃xff0c;这些有用的不同之处使Spark在某些工作负载方面表现得更加优越&＃xff0c;换句话说&＃xff0c;Spark启用了内存分布数据集&＃xff0c;除了能够提供交互式查询外&＃xff0c;它还可以优化迭代工作负载。
Spark是在Scala语言中实现的&＃xff0c;它将Scala用作其应用程序框架。与Hadoop不同&＃xff0c;Spark和Scala能够紧密集成&＃xff0c;其中的Scala可以像操作本地集合对象一样轻松地操作分布式数据集。

尽管创建Spark是为了支持分布式数据集上的迭代作业&＃xff0c;但是实际上它是对Hadoop的补充&＃xff0c;可以在Hadoop文件系统中并行运行。通过名为Mesos的第三方集群框架可以支持此行为。
Spark是一种包含流处理能力的下一代批处理框架,与Hadoop的MapReduce引擎基于各种相同原则开发而来的Spark主要侧重于通过完善的内存计算和处理优化机制加快批处理工作负载的运行速度。Spark可作为独立集群部署&＃xff08;需要相应存储层的配合&＃xff09;&＃xff0c;或可与Hadoop集成并取代MapReduce引擎&＃xff08;也可以与hadoop中MapReduce兼容&＃xff09;MapReduce属于Hadoop生态系统之一&＃xff0c;Spark则属于BDAS&＃xff08;伯克利分析栈&＃xff09;生态系统之一。BDAS包含了Spark&＃xff0c;shark&＃xff08;spark SQL为Shark的升级版&＃xff09;&＃xff0c;BlinkDB&＃xff0c;Spark Streaming

本次采用安装包&＃xff1a;spark-2.3.0-bin-hadoop2.7.tgz

spark-2.3.0的改变&＃xff1a;

1&＃xff09;Spark runs on Java 8&＃43;, Python 2.7&＃43;/3.4&＃43; and R 3.1&＃43;. For the Scala API, Spark 2.3.0 uses Scala 2.11. You will need to use a compatible Scala version (2.11.x)
spark2.3运行环境&＃xff1a;java8&＃xff0c;Python2.7&＃43;/3.4&＃xff0c;R 3.1 Scala2.11
2&＃xff09;Note that support for Java 7, Python 2.6 and old Hadoop versions before 2.6.5 were removed as of Spark 2.2.0. Support for Scala 2.10 was removed as of 2.3.0支持Java7&＃xff0c;Python2.6和hadoop2.6.5之前版本已经在spark2.2移除2.3支持的是scala2.11

3&＃xff09;spark融合的API&＃xff1a;Scala&＃xff0c;Java&＃xff0c;Python&＃xff0c;R&＃xff0c;SQL,Built-in Function

4&＃xff09;spark依赖于&＃xff08;可在运行&＃xff09;&＃xff1a;Mesos&＃xff0c;Yarn&＃xff0c;Kubernetes

spark架构图&＃xff0c;如下图&＃xff1a;

Cluster Manager&＃xff1a;在standalone模式中即为Master主节点&＃xff0c;控制整个集群&＃xff0c;监控worker。在YARN模式中为资源管理器
Worker节点&＃xff1a;从节点&＃xff0c;负责控制计算节点&＃xff0c;启动Executor或者Driver。
Driver&＃xff1a; 运行Application 的main()函数Executor&＃xff1a;执行器&＃xff0c;是为某个Application运行在worker node上的一个进程

spark运行流程图

1.构建Spark Application的运行环境&＃xff0c;启动SparkContext    
2.SparkContext向资源管理器&＃xff08;可以是Standalone&＃xff0c;Mesos&＃xff0c;Yarn&＃xff09;申请运行Executor资源&＃xff0c;并启动StandaloneExecutorbackend    
3.Executor向SparkContext申请Task    
4.SparkContext将应用程序分发给Executor    
5.SparkContext构建成DAG图&＃xff0c;将DAG图分解成Stage、将Taskset发送给Task Scheduler&＃xff0c;最后由Task Scheduler将Task发送给Executor运行    
6.Task在Executor上运行&＃xff0c;运行完释放所有资源

Spark运行特点

1.每个Application获取专属的executor进程&＃xff0c;该进程在Application期间一直驻留&＃xff0c;并以多线程方式运行Task。这种Application隔离机制是有优势的&＃xff0c;无论是从调度角度看&＃xff08;每个Driver调度他自己的任务&＃xff09;&＃xff0c;还是从运行角度看&＃xff08;来自不同Application的Task运行在不同JVM中&＃xff09;&＃xff0c;当然这样意味着Spark Application不能跨应用程序共享数据&＃xff0c;除非将数据写入外部存储系统
2.Spark与资源管理器无关&＃xff0c;只要能够获取executor进程&＃xff0c;并能保持相互通信就可以了
3.提交SparkContext的Client应该靠近Worker节点&＃xff08;运行Executor的节点&＃xff09;&＃xff0c;最好是在同一个Rack里&＃xff0c;因为Spark Application运行过程中SparkContext和Executor之间有大量的信息交换
4.Task采用了数据本地性和推测执行的优化机制

Spark与hadoop&＃xff1a;    
Hadoop有两个核心模块&＃xff0c;分布式存储模块HDFS和分布式计算模块Mapreduce&＃xff1b;spark本身并没有提供分布式文件系统&＃xff0c;因此spark的分析大多依赖于Hadoop的分布式文件系统HDFS Hadoop的Mapreduce与spark都可以进行数据计算&＃xff0c;而相比于Mapreduce&＃xff0c;spark的速度更快并且提供的功能更加丰富。

spark组件、功能介绍

Spark Core&＃xff1a;    
包含Spark的基本功能&＃xff1b;尤其是定义RDD的API、操作以及这两者上的动作。其他Spark的库都是构建在RDD和Spark Core之上的
MLlib&＃xff1a;数据处理和机器学习的算法库    
构建在apache spark之上&＃xff0c;一个专门针对大量数据处理的通用的、快速的引擎。是spark的可以扩展的机器学习库&＃xff0c;由以下部分组成&＃xff1a;通用的学习算法和工具类&＃xff0c;包括分类&＃xff0c;回归&＃xff0c;聚类&＃xff0c;协同过滤&＃xff0c;降维&＃xff0c;当然也包括调优的部分。    
MLlib使用了线性代数包 Breeze, 它依赖于netlib-java和jblas。netlib-java 和 jblas 需要依赖native Fortran routines。    
所以你需要安装安装方法在这个链接中&＃xff1a;https://github.com/mikiobraun/jblas/wiki/Missing-Libraries&＃xff0c;如果你的    
集群的节点中没有安装native Fortran routines。MLlib 会抛出一个link error&＃xff0c;如果没有安装native Fortran routines。    
RDD    
Resilient Distributed Datasets&＃xff1a;弹性分布式数据集&＃xff0c;它具备像MapReduce等数据流模型的容错特性&＃xff0c;并且允许开发人员在    
大型集群上执行基于内存的计算。现有的数据流系统对两种应用的处理并不高效&＃xff1a;一是迭代式算法&＃xff0c;这在图应用和机器学习领域  很常见&＃xff1b;二是交互式数据挖掘工具。这两种情况下&＃xff0c;将数据保存在内存中能够极大地提高性能。为了有效地实现容错&＃xff0c;RDD提供 了一种高度受限的共享内存&＃xff0c;即RDD是只读的&＃xff0c;并且只能通过其他RDD上的批量操作来创建。尽管如此&＃xff0c;RDD仍然足以表示很多类型的计算&＃xff0c;包括MapReduce和专用的迭代编程模型&＃xff08;如Pregel&＃xff09;等。    
RDD是只读的、分区记录的集合&＃xff1a;RDD只能基于在稳定物理存储中的数据集和其他已有的RDD上执行确定性操作来创建。这些确定性操作称之为转换&＃xff0c;如map、filter、groupBy、join&＃xff08;转换不是程开发人员在RDD上执行的操作&＃xff09;    
RDD不需要物化。RDD含有如何从其他RDD衍生&＃xff08;即计算&＃xff09;出本RDD的相关信息&＃xff08;即Lineage&＃xff09;&＃xff0c;据此可以从物理存储的数据计算出相应的RDD分区。    
RDD作为数据结构&＃xff0c;本质上是一个只读的分区记录集合。一个RDD可以包含多个分区&＃xff0c;每个分区就是一个dataset片段。    
RDD可以相互依赖。如果RDD的每个分区最多只能被一个Child RDD的一个分区使用&＃xff0c;则称之为narrow dependency&＃xff08;狭窄依赖&＃xff09;&＃xff1b;    
若多个Child RDD分区都可以依赖&＃xff0c;则称之为wide dependency。不同的操作依据其特性&＃xff0c;可能会产生不同的依赖。例如map操作会产生narrow dependency&＃xff0c;而join操作则产生wide dependency。

GraphX    
GraphX是构建在Spark之上的图计算框架&＃xff0c;它使用RDD来存储图数据&＃xff0c;并提供了实用的图操作方法。由于RDD的与生俱来的特性&＃xff0c;GraphX高效地实现了图的分布式存储和处理&＃xff0c;可以应用于社交网络等大规模的图计算场景。GraphX在图顶点信息和边信息存储上做了优化&＃xff0c;使得图计算框架性能相对于原生RDD实现得以较大提升&＃xff0c;接近或到达GraphLab等专业图计算平台的性能。GraphX最大的贡献是&＃xff0c;在Spark之上提供一栈式数据解决方案&＃xff0c;可以方便且高效地完成图计算的一整套流水作业。    
控制图、并行图操作和计算的一组算法和工具的集合。GraphX扩展了RDD API&＃xff0c;包含控制图、创建子图、访问路径上所有顶点的操作    
Spark SQL(shark升级版)    
Spark SQL is Apache Spark&＃39;s module for working with structured(结构) data SparkSQL的前身是Shark&＃xff0c;给熟悉RDBMS但又不理解MapReduce的技术人员提供快速上手的工具&＃xff0c;Hive应运而生&＃xff0c;它是当时唯一运行在Hadoop上的SQL-on-Hadoop工具。但是MapReduce计算过程中大量的中间磁盘落地过程消耗了大量的I/O&＃xff0c;降低的运行效率&＃xff0c;为了提高SQL-on-Hadoop的效率&＃xff0c;大量的SQL-on-Hadoop工具开始产生&＃xff0c;其中表现较为突出的是&＃xff1a;①MapR的Drill②Cloudera的Impala    
③Shark。其中Shark是伯克利实验室Spark生态环境的组件之一&＃xff0c;它修改了内存管理、物理计划、执行三个模块&＃xff0c;并使之能运行在    
Spark引擎上&＃xff0c;从而使得SQL查询的速度得到10-100倍的提升。    
但是&＃xff0c;随着Spark的发展&＃xff0c;对于野心勃勃的Spark团队来说&＃xff0c;Shark对于Hive的太多依赖&＃xff08;如采用Hive的语法解析器、查询优化器等等&＃xff09;&＃xff0c;制约了Spark的One Stack Rule Them All的既定方针&＃xff0c;制约了Spark各个组件的相互集成&＃xff0c;所以提出了SparkSQL项目。SparkSQL抛弃原有Shark的代码&＃xff0c;汲取了Shark的一些优点&＃xff0c;如内存列存储&＃xff08;In-Memory Columnar Storage&＃xff09;、Hive兼容性等&＃xff0c;重新开发了SparkSQL代码&＃xff1b;由于摆脱了对Hive的依赖性&＃xff0c;SparkSQL无论在数据兼容、性能优化、组件扩展方面都得到了极大的方便&＃xff0c;真可谓“退一步&＃xff0c;海阔天空”。    
提供通过Apache Hive的SQL变体Hive查询语言&＃xff08;HiveQL&＃xff09;与Spark进行交互的API。每个数据库表被当做一个RDD&＃xff0c;Spark SQL查询被转换为Spark操作。    
Shark(类似Hive)    
Spark Streaming    
Spark Streaming 是Spark核心API的一个扩展&＃xff0c;可以实现高吞吐量的、具备容错机制的实时流数据的处理。支持从多种数    
据源获取数据&＃xff0c;包括Kafk、Flume、Twitter、ZeroMQ、Kinesis 以及TCP sockets&＃xff0c;从数据源获取数据之后&＃xff0c;可以使用    
诸如map、reduce、join和window等高级函数进行复杂算法的处理。最后还可以将处理结果存储到文件系统&＃xff0c;数据库和    
现场仪表盘。在“One Stack rule them all”的基础上&＃xff0c;还可以使用Spark的其他子框架&＃xff0c;如集群学习、图计算等&＃xff0c;对    
流数据进行处理。Spark的各个子框架&＃xff0c;都是基于核心Spark的&＃xff0c;Spark Streaming在内部的处理机制是&＃xff0c;接收实时流的数据&＃xff0c;    
并根据一定的时间间隔拆分成一批批的数据&＃xff0c;然后通过Spark Engine处理这些批数据&＃xff0c;最终得到处理后的一批批结果数据。    
对应的批数据&＃xff0c;在Spark内核对应一个RDD实例&＃xff0c;因此&＃xff0c;对应流数据的DStream可以看成是一组RDDs&＃xff0c;即RDD的一个序列。通俗点理解的话&＃xff0c;在流数据分成一批一批后&＃xff0c;通过一个先进先出的队列&＃xff0c;然后 Spark Engine从该队列中依次取出一个个批数据&＃xff0c;把批数据 封装成一个RDD&＃xff0c;然后进行处理&＃xff0c;这是一个典型的生产者消费者模型&＃xff0c;对应的就有生产者消费者模型的问题&＃xff0c;即如何协调生产速率和消费速率。    
对实时数据流进行处理和控制。Spark Streaming允许程序能够像普通RDD一样处理实时数据。

Kafka        
一种高吞吐量的分布式发布订阅消息系统&＃xff0c;它可以处理消费者规模的网站中的所有动作流数据。这种动作&＃xff08;网页浏览&＃xff0c;搜索和其他用户的行动&＃xff09;是在现代网络上的许多社会功能的一个关键因素。 这些数据通常是由于吞吐量的要求而通过处理日志和日志聚合来解决。 对于像Hadoop的一样的日志数据和离线分析系统&＃xff0c;但又要求实时处理的限制&＃xff0c;这是一个可行的解决方案。Kafka的目的是通过Hadoop的并行加载机制来统一线上和离线的消息处理&＃xff0c;也是为了通过集群来提供实时的消费。        
Kafka有如下特下&＃xff1a;        
通过O(1)的磁盘数据结构提供消息的持久化&＃xff0c;这种结构对于即使数以TB的消息存储也能够保持长时间的稳定性能。        
高吞吐量&＃xff1a;即使是非常普通的硬件Kafka也可以支持每秒数百万的消息。支持通过Kafka服务器和消费机集群来分区消息。        
支持Hadoop并行数据加载。

SparkR    
sparkR在spark2.0里面&＃xff0c;RDD后端代码位于org.apache.spark.rdd中&＃xff0c;R语言相关的位于org.apache.spark.api.r中。    
SparkR是一个R语言包&＃xff0c;它提供了轻量级的方式使得可以在R语言中使用Apache Spark。在Spark 1.4中&＃xff0c;SparkR实现了分布式的data frame&＃xff0c;支持类似查询、过滤以及聚合的操作&＃xff08;类似于R中的data frames&＃xff1a;dplyr)&＃xff0c;但是这个可以操作大规模的数据集。使用spark的两种方式&＃xff1a;    
1.在sparkR的shell中交互式使用sparkR    
2.在R脚本中使用    
例&＃xff1a;    if (nchar(Sys.getenv("SPARK_HOME")) <1) {
      Sys.setenv(SPARK_HOME &＃61; "/home/spark")
    }
    library(SparkR, lib.loc &＃61; c(file.path(Sys.getenv("SPARK_HOME"), "R", "lib")))
    sc <- sparkR.init(master &＃61; "spark://10.137",sparkEnvir &＃61; list(spark.driver.memory&＃61;"3g"))

Mesos        
Apache下的开源分布式资源管理框架&＃xff0c;它被称为是分布式系统的内核。        
优点&＃xff1a;        
1.资源管理策略Dominant Resource Fairness(DRF), 这是Mesos的核心&＃xff0c;也是我们把Mesos比作分布式系统Kernel的根本原因。通俗讲&＃xff0c;Mesos能够保证集群内的所有用户有平等的机会使用集群内的资源&＃xff0c;这里的资源包括CPU&＃xff0c;内存&＃xff0c;磁盘等等。很多人拿Mesos跟k8s相比&＃xff0c;我对k8s了解不深&＃xff0c;但是&＃xff0c;我认为这两者侧重点不同不能做比较&＃xff0c;k8s只是负责容器编排而不是集群资源管理。不能因为都可以管理docker&＃xff0c;我们就把它们混为一谈。        
2.轻量级。相对于 Yarn&＃xff0c;Mesos 只负责offer资源给framework&＃xff0c;不负责调度资源。这样&＃xff0c;理论上&＃xff0c;我们可以让各种东西使用Mesos集群资源&＃xff0c;而不像yarn只拘泥于hadoop&＃xff0c;我们需要做的是开发调度器&＃xff08;mesos framework&＃xff09;。        
3.提高分布式集群的资源利用率&＃xff1a;这是一个 generic 的优点。从某些方面来说&＃xff0c;所有的集群管理工具都是为了提高资源利用率。VM的出现&＃xff0c;催生了IaaS&＃xff1b;容器的出现&＃xff0c;催生了k8s, Mesos等等。简单讲&＃xff0c;同样多的资源&＃xff0c;我们利用IaaS把它们拆成VM 与 利用k8s/Mesos把它们拆成容器&＃xff0c;显然后者的资源利用率更高。(这里我没有讨论安全的问题&＃xff0c;我们假设内部子网环境不需要考虑这个。)
缺点&＃xff1a;        
1.门槛太高。只部署一套Mesos&＃xff0c;你啥都干不了&＃xff0c;为了使用它&＃xff0c;你需要不同的mesos framework&＃xff0c;像Marathon&＃xff0c;chronos&＃xff0c;spark等等。或者自己写framework来调度Mesos给的资源&＃xff0c;这让大家望而却步。        
2.目前对stateful service的支持不够。Mesos集群目前无法进行数据持久化。即将发布的0.23版本增加了persistent resource和dynamic reserver&＃xff0c;数据持久化问题将得到改善。        
3.脏活累活不会少。Team在使用Mesos前期很乐观&＃xff0c;认为搞定了Mesos&＃xff0c;我们的运维同学能轻松很多。然而&＃xff0c;根本不是那么回事儿&＃xff0c;集群节点的优化&＃xff0c;磁盘&＃xff0c;网络的设置&＃xff0c;等等这些&＃xff0c;Mesos是不会帮你干的。使用初期&＃xff0c;运维的工作量不仅没有减轻&＃xff0c;反而更重了。        
4.Mesos项目还在紧锣密鼓的开发中&＃xff0c;很多功能还不完善。譬如&＃xff0c;集群资源抢占还不支持。        
Kubernetes        
Kubernetes&＃xff08;k8s&＃xff09;是自动化容器操作的开源平台&＃xff0c;这些操作包括部署&＃xff0c;调度和节点集群间扩展。如果你曾经用过Docker容器技术部署容器&＃xff0c;那么可以将Docker看成Kubernetes内部使用的低级别组件。Kubernetes不仅仅支持Docker&＃xff0c;还支持Rocket&＃xff0c;这是另一种容器技术。        
使用Kubernetes可以&＃xff1a;        
    自动化容器的部署和复制    
    随时扩展或收缩容器规模    
    将容器组织成组&＃xff0c;并且提供容器间的负载均衡    
    很容易地升级应用程序容器的新版本    
    提供容器弹性&＃xff0c;如果容器失效就替换它&＃xff0c;等等...

yarn    
Apache Hadoop YARN &＃xff08;Yet Another Resource Negotiator&＃xff0c;另一种资源协调者&＃xff09;是一种新的 Hadoop 资源管理器&＃xff0c;它是一个通用资源管理系统&＃xff0c;可为上层应用提供统一的资源管理和调度&＃xff0c;它的引入为集群在利用率、资源统一管理和数据共享等方面带来了巨大好处。    
Yarn的优点&＃xff1a;    
1.大大减小了 JobTracker&＃xff08;也就是现在的 ResourceManager&＃xff09;的资源消耗&＃xff0c;并且让监测每一个 Job 子任务 (tasks) 状态的程序分布式化了&＃xff0c;更安全、更优美。    
2.在新的 Yarn 中&＃xff0c;ApplicationMaster 是一个可变更的部分&＃xff0c;用户可以对不同的编程模型写自己的 AppMst&＃xff0c;让更多类型的编程模型能够跑在 Hadoop 集群中&＃xff0c;可以参考 hadoop Yarn 官方配置模板中的 mapred-site.xml 配置。    
3.对于资源的表示以内存为单位 ( 在目前版本的 Yarn 中&＃xff0c;没有考虑 cpu 的占用 )&＃xff0c;比之前以剩余 slot 数目更合理。    
4.老的框架中&＃xff0c;JobTracker 一个很大的负担就是监控 job 下的 tasks 的运行状况&＃xff0c;现在&＃xff0c;这个部分就扔给     
ApplicationMaster 做了&＃xff0c;而 ResourceManager 中有一个模块叫做 ApplicationsMasters( 注意不是 ApplicationMaster)&＃xff0c;    
它是监测 ApplicationMaster 的运行状况&＃xff0c;如果出问题&＃xff0c;会将其在其他机器上重启。    
5.Container 是 Yarn 为了将来作资源隔离而提出的一个框架。这一点应该借鉴了 Mesos 的工作&＃xff0c;目前是一个框架&＃xff0c;仅仅提供    
java 虚拟机内存的隔离,hadoop 团队的设计思路应该后续能支持更多的资源调度和控制,既然资源表示成内存量&＃xff0c;那就没    
有了之前的 map slot/reduce slot 分开造成集群资源闲置的尴尬情况。    
YARN的核心思想&＃xff1a;    
将JobTracker和TaskTracker进行分离&＃xff0c;它由下面几大构成组件&＃xff1a;    
    a. 一个全局的资源管理器 ResourceManager
    b.ResourceManager的每个节点代理 NodeManager
    c. 表示每个应用的 ApplicationMaster
    d. 每一个ApplicationMaster拥有多个Container在NodeManager上运行

Spark常用术语

1)Application        
Appliction都是指用户编写的Spark应用程序&＃xff0c;其中包括一个Driver功能的代码和分布在集群中多个节点上运行的Executor代码        
2)Driver        
Spark中的Driver即运行上述Application的main函数并创建SparkContext&＃xff0c;创建SparkContext的目的是为了准备Spark应用程序的运行环境&＃xff0c;在Spark中有SparkContext负责与ClusterManager通信&＃xff0c;进行资源申请、任务的分配和监控等&＃xff0c;当Executor部分运行完毕后&＃xff0c;Driver同时负责将SparkContext关闭&＃xff0c;通常用SparkContext代表Driver        
3)Executor        
某个Application运行在worker节点上的一个进程&＃xff0c;  该进程负责运行某些Task&＃xff0c; 并且负责将数据存到内存或磁盘上&＃xff0c;每个Application都有各自独立的一批Executor&＃xff0c; 在Spark on Yarn模式下&＃xff0c;其进程名称为CoarseGrainedExecutor Backend。一个CoarseGrainedExecutor Backend有且仅有一个Executor对象&＃xff0c; 负责将Task包装成taskRunner,并从线程池中抽取一个空闲线程运行Task&＃xff0c; 这个每一个CoarseGrainedExecutor Backend能并行运行Task的数量取决于分配给它的cpu个数        
4)Cluter Manager        
指的是在集群上获取资源的外部服务。目前有三种类型:        
    Standalon : spark原生的资源管理&＃xff0c;由Master负责资源的分配    
    Apache Mesos:与hadoop MR兼容性良好的一种资源调度框架    
    Hadoop Yarn: 主要是指Yarn中的ResourceManager

5)Worker    
集群中任何可以运行Application代码的节点&＃xff0c;在Standalone模式中指的是通过slave文件配置的Worker    
节点&＃xff0c;在Spark on Yarn模式下就是NoteManager节点    
6)Task    
被送到某个Executor上的工作单元&＃xff0c;但hadoopMR中的MapTask和ReduceTask概念一样&＃xff0c;是运行Application    
的基本单位&＃xff0c;多个Task组成一个Stage&＃xff0c;而Task的调度和管理等是由TaskScheduler负责    
7)Job    
包含多个Task组成的并行计算&＃xff0c;往往由Spark Action触发生成&＃xff0c; 一个Application中往往会产生多个Job    
8)Stage    
每个Job会被拆分成多组Task&＃xff0c; 作为一个TaskSet&＃xff0c; 其名称为Stage&＃xff0c;Stage的划分和调度是有DAGScheduler    
来负责的&＃xff0c;Stage有非最终的Stage&＃xff08;Shuffle Map Stage&＃xff09;和最终的Stage&＃xff08;Result Stage&＃xff09;两种&＃xff0c;Stage的    
边界就是发生shuffle的地方    
9)DAGScheduler   

根据Job构建基于Stage的DAG                    
&＃xff08;Directed Acyclic Graph有向无环图)&＃xff0c;    
并提交Stage给TASkScheduler。 其划分    
Stage的依据是RDD之间的依赖的关系找出    
开销最小的调度方法

10)TASKSedulter

将TaskSET提交给worker运行&＃xff0c;每个Executor运行什么Task就是在此处分配的. TaskScheduler维护所有TaskSet&＃xff0c;当Executor向Driver发生心跳时TaskScheduler会根据资源剩余情况分配相应的Task。另外&＃xff0c;TaskScheduler还维护着所有Task的运行标签&＃xff0c;重试失败的Task。

11)在不同运行模式中任务调度器具体为&＃xff1a;    
    Spark on Standalone模式为TaskScheduler
    YARN-Client模式为YarnClientClusterScheduler
    YARN-Cluster模式为YarnClusterScheduler

术语总结&＃xff1a;    
Job&＃61;多个stage&＃xff0c;Stage&＃61;多个同种task    Task分为ShuffleMapTask和ResultTask    
Dependency分为&＃xff1a;    
    ShuffleDependency
    NarrowDependency

Spark运行模式

概述    
Spark的运行模式多种多样&＃xff0c;灵活多变&＃xff0c;部署在单机上时&＃xff0c;既可以用本地模式运行&＃xff0c;也可以用伪分布模式运行&＃xff0c;而当以分布式集群的方式部署时&＃xff0c;也有众多的运行模式可供选择&＃xff0c;这取决于集群的实际情况&＃xff0c;底层的资源调度即可以依赖外部资源调度框架&＃xff0c;也可以使用Spark内建的Standalone模式。    
对于外部资源调度框架的支持&＃xff0c;目前的实现包括相对稳定的Mesos模式&＃xff0c;以及hadoop YARN模式    
本地模式&＃xff1a;常用于本地开发测试&＃xff0c;本地还分别 local 和 local cluster

1)standalone: 独立集群运行模式    
Standalone模式使用Spark自带的资源调度框架采用Master/Slaves的典型架构&＃xff0c;选用ZooKeeper来实现Master的HA

框架结构图如下&＃xff1a;

该模式主要的节点有Client节点、Master节点和Worker节点。其中Driver既可以运行在Master节点上中&＃xff0c;也可以运行在本地Client端。当用spark-shell交互式工具提交Spark的Job时&＃xff0c;Driver在Master节点上运行&＃xff1b;当使用spark-submit工具提交Job或者在Eclips、IDEA等开发平台上使用”new SparkConf.setManager(“spark://master:7077”)”方式运行Spark任务时&＃xff0c;Driver是运行在本地Client端上的。

运行过程如下图&＃xff1a;

1.SparkContext连接到Master&＃xff0c;向Master注册并申请资源&＃xff08;CPU Core 和Memory&＃xff09;
2.Master根据SparkContext的资源申请要求和Worker心跳周期内报告的信息决定在哪个Worker上分配资源&＃xff0c;然后在该Worker上获取资源&＃xff0c;然后启动StandaloneExecutorBackend
3.StandaloneExecutorBackend向SparkContext注册&＃xff1b;

4.SparkContext将Applicaiton代码发送给StandaloneExecutorBackend&＃xff1b;并且SparkContext解析Applicaiton代码&＃xff0c;构建DAG图&＃xff0c;并提交给DAG Scheduler分解成Stage&＃xff08;当碰到Action操作时&＃xff0c;就会催生Job&＃xff1b;每个Job中含有1个或多个Stage&＃xff0c;Stage一般在获取外部数据和shuffle之前产生&＃xff09;&＃xff0c;然后以Stage&＃xff08;或者称为TaskSet&＃xff09;提交给Task Scheduler&＃xff0c;Task Scheduler负责将Task分配到相应的Worker&＃xff0c;最后提交给StandaloneExecutorBackend执行&＃xff1b;
5.StandaloneExecutorBackend会建立Executor线程池&＃xff0c;开始执行Task&＃xff0c;并向SparkContext报告&＃xff0c;直至Task完成
6.所有Task完成后&＃xff0c;SparkContext向Master注销&＃xff0c;释放资源

2)Spark on YARN&＃xff1a;在yarn上运行    
Spark on YARN模式根据Driver在集群中的位置分为两种模式&＃xff1a;    
    YARN-Client模式
    YARN-Cluster&＃xff08;或称为YARN-Standalone模式&＃xff09;
A.Yarn-Client模式中&＃xff1a;    
Driver在客户端本地运行&＃xff0c;这种模式可以使得Spark Application和客户端进行交互&＃xff0c;因为Driver在客户端&＃xff0c;所以可以通过webUI访问Driver的状态&＃xff0c;默认是http://hadoop1:4040访问&＃xff0c;而YARN通过http://hadoop1:8088访问    
YARN-client的工作流程步骤为:

a.Spark Yarn Client向YARN的ResourceManager申请启动Application Master。同时在SparkContent        
初始化中将创建DAGScheduler和TASKScheduler等&＃xff0c;由于我们选择的是Yarn-Client模式&＃xff0c;程序会选择        
YarnClientClusterScheduler和YarnClientSchedulerBackend        
b.ResourceManager收到请求后&＃xff0c;在集群中选择一个NodeManager&＃xff0c;为该应用程序分配第一个Container&＃xff0c;        
要求它在这个Container中启动应用程序的ApplicationMaster&＃xff0c;与YARN-Cluster区别的是在该        
ApplicationMaster不运行SparkContext&＃xff0c;只与SparkContext进行联系进行资源的分派        
c.Client中的SparkContext初始化完毕后&＃xff0c;与ApplicationMaster建立通讯&＃xff0c;向ResourceManager        
注册&＃xff0c;根据任务信息向ResourceManager申请资源&＃xff08;Container&＃xff09;        
d.一旦ApplicationMaster申请到资源&＃xff08;也就是Container&＃xff09;后&＃xff0c;便与对应的NodeManager通信&＃xff0c;要求        
它在获得的Container中启动CoarseGrainedExecutorBackend&＃xff0c;CoarseGrainedExecutorBackend启        
动后会向Client中的SparkContext注册并申请Task        
e.client中的SparkContext分配Task给CoarseGrainedExecutorBackend执行&＃xff0c;CoarseGrainedExecutorBackend        
运行Task并向Driver汇报运行的状态和进度&＃xff0c;以让Client随时掌握各个任务的运行状态&＃xff0c;从而可以在任务        
失败时重新启动任务        
f.应用程序运行完成后&＃xff0c;Client的SparkContext向ResourceManager申请注销并关闭自己        
B.Spark Cluster模式        
在YARN-Cluster模式中&＃xff0c;当用户向YARN中提交一个应用程序后&＃xff0c;YARN将分两个阶段运行该应用程序&＃xff1a;        
    第一个阶段是把Spark的Driver作为一个ApplicationMaster在YARN集群中先启动&＃xff1b;    
    第二个阶段是由ApplicationMaster创建应用程序&＃xff0c;然后为它向ResourceManager申请资源&＃xff0c;    
    并启动Executor来运行Task&＃xff0c;同时监控它的整个运行过程&＃xff0c;直到运行完成

YARN-cluster的工作流程分为以下几个步骤:

a.Spark Yarn Client向YARN中提交应用程序&＃xff0c;包括ApplicationMaster程序、启动ApplicationMaster的命令、需要在Executor中运行的程序等        
b.ResourceManager收到请求后&＃xff0c;在集群中选择一个NodeManager&＃xff0c;为该应用程序分配第一个Container&＃xff0c;要求它在这个Container中启动应用程序的ApplicationMaster&＃xff0c;其中ApplicationMaster进行SparkContext等的初始化        
c.ApplicationMaster向ResourceManager注册&＃xff0c;这样用户可以直接通过ResourceManage查看应用程序的运行状态&＃xff0c;然后它将采用轮询的方式通过RPC协议为各个任务申请资源&＃xff0c;并监控它们的运行状态直到运行结束        
d.一旦ApplicationMaster申请到资源&＃xff08;也就是Container&＃xff09;后&＃xff0c;便与对应的NodeManager通信&＃xff0c;要求它在        
获得的Container中启动CoarseGrainedExecutorBackend&＃xff0c;CoarseGrainedExecutorBackend启动后会向ApplicationMaster中的SparkContext注册并申请Task。这一点和Standalone模式一样&＃xff0c;只不过SparkContext在Spark Application中初始化时&＃xff0c;使用CoarseGrainedSchedulerBackend配合YarnClusterScheduler进行任务的调度&＃xff0c;其中YarnClusterScheduler只是对TaskSchedulerImpl的一个简单包装&＃xff0c;增加了对Executor的等待逻辑等        
e.ApplicationMaster中的SparkContext分配Task给CoarseGrainedExecutorBackend执行&＃xff0c;CoarseGrainedExecutorBackend运行Task并向ApplicationMaster汇报运行的状态和进度&＃xff0c;以让ApplicationMaster随时掌握各个任务的运行状态&＃xff0c;从而可以在任务失败时重新启动任务        
f.应用程序运行完成后&＃xff0c;ApplicationMaster向ResourceManager申请注销并关闭自己

Spark Client 和 Spark Cluster的区别:        
理解YARN-Client和YARN-Cluster深层次的区别之前先清楚一个概念&＃xff1a;Application Master。在YARN中&＃xff0c;每个Application实例都有一个ApplicationMaster进程&＃xff0c;它是Application启动的第一个容器。它负责和ResourceManager打交道并请求资源&＃xff0c;获取资源之后告诉NodeManager为其启动Container。从深层次的含义讲YARN-Cluster和YARN-Client模式的区别其实就是ApplicationMaster进程的区别YARN-Cluster模式下&＃xff0c;Driver运行在AM(Application Master)中&＃xff0c;它负责向YARN申请资源&＃xff0c;并监督作业的运行状况。当用户提交了作业之后&＃xff0c;就可以关掉Client&＃xff0c;作业会继续在YARN上运行&＃xff0c;因而    
YARN-Cluster模式不适合运行交互类型的作业        
YARN-Client模式下&＃xff0c;Application Master仅仅向YARN请求Executor&＃xff0c;Client会和请求的Container通信来调度他们工作&＃xff0c;也就是说Client不能离开

RDD运行流程

RDD在Spark中运行大概分为以下三步&＃xff1a;        
    1.创建RDD对象    
    2.DAGScheduler模块介入运算&＃xff0c;计算RDD之间的依赖关系&＃xff0c;RDD之间的依赖关系就形成了DAG    
    3.每一个Job被分为多个Stage。划分Stage的一个主要依据是当前计算因子的输入是否是确定的&＃xff0c;    
    如果是则将其分在同一个Stage&＃xff0c;避免多个Stage之间的消息传递开销

以下面一个按 A-Z 首字母分类&＃xff0c;查找相同首字母下不同姓名总个数的例子来看一下 RDD 是如何运行起来的

创建 RDD  上面的例子除去最后一个 collect 是个动作&＃xff0c;不会创建 RDD 之外&＃xff0c;前面四个转换都
会创建出新的 RDD 。因此第一步就是创建好所有 RDD( 内部的五项信息 )&＃xff1f;
创建执行计划 Spark 会尽可能地管道化&＃xff0c;并基于是否要重新组织数据来划分 阶段 (stage) &＃xff0c;
例如本例中的 groupBy() 转换就会将整个执行计划划分成两阶段执行。最终会产生一个
DAG(directed acyclic graph &＃xff0c;有向无环图 ) 作为逻辑执行计划

调度任务将各阶段划分成不同的任务 (task) &＃xff0c;每个任务都是数据和计算的合体。在进行下一阶段前&＃xff0c;当前阶段的所有任务都要执行完成。因为下一阶段的第一个转换一定是重新组织数据的&＃xff0c;所以必须等当前阶段所有结果数据都计算出来了才能继续

Spark安装配置

1)Scala定义        
Scala是一门多范式&＃xff08;multi-paradigm&＃xff09;的编程语言&＃xff0c;设计初衷是要集成面向对象编程和函数式编程的各种特性。        
Scala 运行在Java虚拟机上&＃xff0c;并兼容现有的Java程序。Scala 源代码被编译成Java字节码&＃xff0c;所以它可以运行于JVM之上&＃xff0c;并可以调用现有的Java类库。
2&＃xff09;安装scala        
因为Spark的配置依赖与Scala&＃xff0c;所以先要配置Scala。把scala-2.12.5.tgz移动到/usr/software目录下        
tar -xvf scala-2.12.5.tgz        
解压之后出现目录&＃xff1a;scala-2.12.5        
配置scala环境变量&＃xff08;每台都装&＃xff09;        
vim /etc/profile        
export SCALA_HOME&＃61;/usr/software/scala-2.12.5        
export PATH&＃61;${SCALA_HOME}/bin:$PATH        
source profile        
验证scala是否安装成功&＃xff1a;scala -version        
Scala code runner version 2.12.5 -- Copyright 2002-2018, LAMP/EPFL and Lightbend, Inc.        
配置scala开机自启&＃xff1a;vim ~/.bashrc        
JAVA_HOME&＃61;/usr/software/java/jdk1.8.0_152        
HADOOP_HOME&＃61;/usr/software/hadoop-3.0.1        
SCALA_HOME&＃61;/usr/software/scala-2.12.5        
CLASSPATH&＃61;.:$JAVA_HOME/lib/dt.jar:$JAVA_HOME/lib/tools.jar        
PATH&＃61;$JAVA_HOME/bin:$HADOOP_HOME/bin:$SCALA_HOME/bin:$PATH        
export JAVA_HOME HADOOP_HOME SCALA_HOME CLASSPATH PATH        
保存退出        
把scala-2.12.5移动到其它机器&＃xff1a;        
scp -r /usr/software/scala-2.12.5 hadoop02&＃64;hadoop02:/home/hadoop02        
将scala-2.12.5移动到/usr/software目录下copy完之后在每台机器上配置环境变量&＃xff1a;/etc/profile和.bashrc两个文件

3)安装配置spark        
离线下载安装包&＃xff1a;spark-2.3.0-bin-hadoop2.7.tgz        
tar -xvf spark-2.3.0-bin-hadoop2.7.tgz        
出现目录&＃xff1a;spark-2.3.0-bin-hadoop2.7        
目录重命名&＃xff1a;mv spark-2.3.0-bin-hadoop2.7 /usr/software/spark        
配置环境变量&＃xff08;每台机器都需要配置&＃xff09;        
vim /etc/profile        
export SPARK_HOME&＃61;/usr/software/spark        
export PATH&＃61;${SPARK_HOME}/bin:$PATH        
source /etc/profile

配置spark-env.sh        
cd /usr/software/spark/conf        
cp spark-env.sh.template spark-env.sh        
vim spark-env.sh        
export JAVA_HOME&＃61;/usr/software/java/jdk1.8.0_152        
export SCALA_HOME&＃61;/usr/software/scala-2.12.5        
export HADOOP_HOME&＃61;/usr/software/hadoop-3.0.1        
export HADOOP_CONF_DIR&＃61;$HADOOP_HOME/etc/hadoop        
export SPARK_HOME&＃61;/usr/software/spark        
export SPARK_MASTER_IP&＃61;hadoop01        
export SPARK_EXECUTOR_MEMORY&＃61;2G        
export SPARK_MASTER_WEBUI_PORT&＃61;8085        
保存退出        
配置$SPARK_HOME/conf/slaves        
cd /usr/software/spark/conf        
cp slaves.template slaves        
vim slaves        
hadoop01        
hadoop02        
hadoop03        
hadoop04        
hadoop05        
保存退出        
把配置好的spark复制到其它节点

scp -r /usr/software/spark hadoop02&＃64;hadoop02:/home/hadoop02            
将spark移动到/usr/software目录下&＃xff0c;配置spark环境变量&＃xff1a;vim /etc/profile

4)启动spark            
启动spark要确保hadoop已经成功启动&＃xff01;&＃xff01;&＃xff01;            
用jps确认            
cd /usr/software/spark            
sbin/start-all.sh            sbin/stop-all.sh

启动后&＃xff0c;jps查看&＃xff0c;hadoop01作为主机多出两个进程&＃xff0c;其它机器多出一个进程

Spark简易测试&＃xff1a;

输入URL&＃xff1a;http://hadoop01:8085/&＃xff0c;查看webUI&＃xff1a;

spark简易测试&＃xff1a;    
为了避免麻烦这里我们使用spark-shell&＃xff0c;做一个简单的worcount的测试    
用于在测试hadoop的时候我们已经在hdfs上存储了测试的源文件&＃xff0c;下面就是直接拿来用就好了&＃xff01;    
cd /usr/software/spark/bin    
spark-shell    
Spark的shell作为一个强大的交互式数据分析工具&＃xff0c;提供了一个简单的方式学习API。它可以使用    
Scala(在Java虚拟机上运行现有的Java库的一个很好方式)或Python。    
在scala>后输入

val file&＃61;sc.textFile("hdfs://mycluster1/input/words.txt")    
val rdd &＃61; file.flatMap(line &＃61;> line.split(" ")).map(word &＃61;> (word,1)).reduceByKey(_&＃43;_)    
rdd.collect()    
rdd.foreach(println)

测试成功后录入&＃xff1a;          :quit