二者都使用mr模型来进行并行计算，hadoop的一个做业称为job，job里面分为map task和reduce task，每一个task都是在本身的进程中运行的，当task结束时，进程也会结束。web
Spark用户提交的任务称为application，一个application对应一个SparkContext，app中存在多个job，没触发一个action操做就会产生一个job。算法
这些job能够并行或者串行执行，每一个job有多个stage，stage是shuffle过程当中DAGSchaduler经过RDD之间的依赖关系划分job而来的，每一个stage里面有多个task，组成taskset有TaskSchaduler分发到各个executor中执行，executor的生命周期是和application同样的，即便没有job运行也是存在的，因此task能够快速启动读取内存进行计算的。sql
Hadoop的job只有map和reduce操做，表达能力比较欠缺并且在mr过程当中会重复的读写hdfs，形成大量的io操做，多个job须要本身管理关系。数据库
Spark的迭代计算都是在内存中进行的，API中提供了大量的RDD操做join，groupby等，并且经过DAG图能够实现良好的容错。编程
Hadoop：map端保存分片数据，经过网络收集到reduce端。
Spark：spark的shuffle实在DAGSchedular划分Stage的时候产生的，TaskSchedular要分发Stage到各个worker的executor。减小shuffle能够提升性能。数组
Spark 有不少种模式，最简单就是单机本地模式，还有单机伪分布式模式，复杂的则运行在集群中，目前能很好的运行在 Yarn和 Mesos 中，固然 Spark 还有自带的 Standalone 模式，对于大多数状况 Standalone 模式就足够了，若是企业已经有 Yarn 或者 Mesos 环境，也是很方便部署的。缓存
standalone(集群模式)：典型的Mater/slave模式，不过也能看出Master是有单点故障的；Spark支持ZooKeeper来实现 HA性能优化
on yarn(集群模式)： 运行在 yarn 资源管理器框架之上，由 yarn 负责资源管理，Spark 负责任务调度和计算网络
on mesos(集群模式)： 运行在 mesos 资源管理器框架之上，由 mesos 负责资源管理，Spark负责任务调度和计算。app
on cloud(集群模式)：好比 AWS 的 EC2，使用这个模式能很方便的访问 Amazon的 S3。Spark支持多种分布式存储系统：HDFS和S3
reduceByKey：reduceByKey会在结果发送至reducer以前会对每一个mapper在本地进行merge，有点相似于在MapReduce中的combiner。这样作的好处在于，在map端进行一次reduce以后，数据量会大幅度减少，从而减少传输，保证reduce端可以更快的进行结果计算。
groupByKey：groupByKey会对每个RDD中的value值进行聚合造成一个序列(Iterator)，此操做发生在reduce端，因此势必会将全部的数据经过网络进行传输，形成没必要要的浪费。同时若是数据量十分大，可能还会形成OutOfMemoryError。
经过以上对比能够发如今进行大量数据的reduce操做时候建议使用reduceByKey。不只能够提升速度，仍是能够防止使用groupByKey形成的内存溢出问题。
Cache后能够接其余算子，可是接了算子以后，起不到缓存的做用，由于会重复出发cache。
Cache不是action操做。
ReduceByKey是transform算子，reduce是action算子
DAG在划分stage时肯定。
lazy记录了数据的来源，RDD是不可变的，且是lazy级别的，且rDD之间构成了链条，lazy是弹性的基石。因为RDD不可变，因此每次操做就产生新的rdd，不存在全局修改的问题，控制难度降低，全部有计算链条将复杂计算链条存储下来，计算的时候从后往前回溯900步是上一个stage的结束，要么就checkpoint
记录原数据，是每次修改都记录，代价很大若是修改一个集合，代价就很小，官方说rdd是粗粒度的操做，是为了效率，为了简化，每次都是操做数据集合，写或者修改操做，都是基于集合的rdd的写操做是粗粒度的，rdd的读操做既能够是粗粒度的也能够是细粒度，读能够读其中的一条条的记录。
简化复杂度，是高效率的一方面，写的粗粒度限制了使用场景如网络爬虫，现实世界中，大多数写是粗粒度的场景
前提是定位数据倾斜，是OOM仍是任务执行缓慢，查看日志或者看4040 web url
解决方法：
避免没必要要的shuffle，如使用广播小表的方式，将reduce-side-join提高为map-side-join
Spark并行度，每一个core承载24个partition，32core，那么64128之间的并行度，也就是设置64~128个partition，并行度和数据规模无关，只和内存使用量和cpu使用时间有关。
每一个数据分片都对应具体物理位置，数据位置是由blockManager管理，不管数据是在磁盘，内存仍是tacyan，都由blockManager管理。
Spark的数据本地性有三种：
PROCESS_LOCAL是指读取缓存在本地节点的数据
NODE_LOCAL是指读取本地节点磁盘的数据
ANY是指读取非本地节点的数据
一般读取数据PROCESS_LOCAL>NODE_LOCAL>ANY，尽可能使数据以PROCESS_LOCAL或NODE_LOCAL方式读取。其中PROCESS_LOCAL还和cache有关，若是RDD常常用的话将该RDD cache到内存中，注意，因为cache是lazy的，因此必须经过一个action的触发，才能真正的将该RDD cache到内存中。
将不能序列化的内容封装成object
Driver将集群中各节点的内容收集起来汇总成结果，collect返回的类型为Array，collect把各个节点上的数据抓过来，抓过来数据是Array类型，collect对Array抓过来的结果进行合并，合并后Array中只有一个元素，是tuple类型(KV)的。
输入数据有不少task，尤为是有不少小文件的时候，有多少个输入block就有多少个task
Spark中有partition的概念。每一个partition对应一个task，task越多，在处理大规模数据的时候，就会越有效率。不过task并非越多越好，若是数据量不大，则不必启动太多task；
参数能够经过spark_home/conf/spark-default.conf配置文件设置:
spark.sql.shuffle.partitions 50 spark.default.parallelism 10
第一个是针对spark sql的task数量
第二个是非spark sql程序设置生效
会致使执行该job时候集群资源不足，致使执行job结束也没有分配足够的资源，分配了部分Executor，该job就开始执行task，应该是task的调度线程和Executor资源申请是异步的。
Map：对RDD每一个元素转换，文件中每一行的数据返回一个数组对象。
Flatmap:对RDD每一个元素转换，而后再扁平化将全部对象合并成一个对象，文件中的全部行数据金返回一个数组对象，会抛弃为null的值。
Spark全部复杂的算法都会有persist的身影，spark默认数据放在内存，spark不少内容都是放在内存的，很是适合高速迭代，1000个步骤，只有第一个输入数据，中间不产生临时数据，但分布式系统风险很高，因此容易出错，就要容错，rdd出错或者分片能够根据血统算出来，若是没有对父rdd进行persist或者cache的话，就须要重头作。
如下场景要进行persist：
某个步骤计算很是耗时，须要进行persist持久化
计算链条很是长，从新恢复要算不少步骤
Checkpoint所在的rdd要持久化persist，lazy级别，框架发现有checkpoint时单独触发一个job，须要从新算一遍，checkpoint前要持久化，写个rdd.cache或者rdd.persist，将结果保存起来，再写checkpoint操做，这样执行起来会很是快，不须要从新计算rdd链条了。
Shuffle以后要persist，shuffle要进行网络传输，风险很大，数据丢失重来，恢复代价很大
Shuffle以前要persist，框架默认将数据持久化到磁盘，这个是框架自动作的
序列化能够减小数据的体积，减小存储空间，高效存储和传输数据，很差的是使用的时候要反序列化，很是消耗cpu。
join其实常见的就分为两类： map-side join 和 reduce-side join。当大表和小表join时，用map-side join能显著提升效率。将多份数据进行关联是数据处理过程当中很是广泛的用法，不过在分布式计算系统中，这个问题每每会变的很是麻烦，由于框架提供的 join 操做通常会将全部数据根据 key 发送到全部的 reduce 分区中去，也就是 shuffle 的过程。形成大量的网络以及磁盘IO消耗，运行效率极其低下，这个过程通常被称为 reduce-side-join。若是其中有张表较小的话，咱们则能够本身实如今 map 端实现数据关联，跳过大量数据进行 shuffle 的过程，运行时间获得大量缩短，根据不一样数据可能会有几倍到数十倍的性能提高。
一个实时毫秒，一个准实时亚秒，不过storm的吞吐率比较低。
Master：管理集群和节点，不参与计算。
Worker：计算节点，进程自己不参与计算，和master汇报。
Driver：运行程序的main方法，建立sparkcontext对象。
Spark context：控制整个application的生命周期，包括DAGSchedular和TaskSchedular等组件。
Client：用户提交程序的入口。
RDD和他依赖的父RDD的关系有两种不一样的类型，即窄依赖(narrow dependency)和宽依赖(wide dependency)。
宽依赖：指的是多个子RDD的partition会依赖同一个父RDD的partition。
窄依赖：指的是每个父RDD的partition最多被子RDD的一个Partition使用。
1.spark application中能够由于不一样的action触发众多的job，一个Application中能够有不少job，每一个job是有一个或多个stage构成的，后面的stage依赖于前面的stage，也就是说只有前面的stage计算完毕后，后面的stage才会运行。
2.stage划分的依据是宽依赖，什么时候产生宽依赖，例如ReduceBykey，GroupByKey的算子，会致使宽依赖的产生。
3.由Action算子(例如collect)致使了SparkContext.RunJob的执行，最终致使了DAGSchedular的submitJob的执行，其核心是经过发送一个case class Jobsubmitted对象给eventProcessLoop。
EventProcessLoop是DAGSchedularEventProcessLoop的具体事例，而DAGSchedularEventProcessLoop是eventLoop的子类，具体实现EventLoop的onReceiver方法，onReceiver方法转过来回调doOnReceive。
4.在handleJobSubmitted中首先建立finalStage，建立finalStage时候会创建父Stage的依赖链条。
总结：依赖是从代码的逻辑层面上来展开说的，能够简单点说：写介绍什么是RDD中的宽窄依赖，而后再根据DAG有向无环图进行划分，从当前job的最后一个算子往前推，遇到宽依赖，那么当前在这个批次中的全部算子操做都划分红一个stage，而后继续按照这种方式再继续往前推，如再遇到宽依赖，又划分红一个stage，一直到最前面的一个算子。最后整个job会被划分红多个stage，而stage之间又存在依赖关系，后面的stage依赖于前面的stage。
在经过spark-submit提交任务时，能够经过添加配置参数来指定：
—driver-class-path 外部jar包
—jars 外部jar包
Cache：缓存数据，默认是缓存在内存中，，其本质仍是调用persist
Persist：缓存数据，有丰富的缓存策略。数据能够保存在内存也能够保存在磁盘中，使用的时候指定对应的缓存级别。
保存元数据，包括流式应用的配置、流式没崩溃以前定义的各类操做、未完成全部操做的batch。元数据被存储到容忍失败的存储系统上，如HDFS。这种ckeckpoint主要针对driver失败后的修复。
保存流式数据，也是存储到容忍失败的存储系统上，如HDFS。这种ckeckpoint主要针对window operation、有状态的操做。不管是driver失败了，仍是worker失败了，这种checkpoint都够快速恢复，而不须要将很长的历史数据都从新计算一遍（以便获得当前的状态）。
对于一个须要作checkpoint的DStream结构，能够经过调用DStream.checkpoint(checkpointInterval)来设置ckeckpoint的周期，经验上通常将这个checkpoint周期设置成batch周期的5至10倍。
这是一个可选功能，建议加上。这个功能将使得输入数据写入以前配置的checkpoint目录。这样有状态的数据能够从上一个checkpoint开始计算。开启的方法是把spark.streaming.receiver.writeAheadLogs.enable这个property设置为true。另外，因为输入RDD的默认StorageLevel是MEMORY_AND_DISK_2，即数据会在两台worker上作replication。实际上，Spark Streaming模式下，任何从网络输入数据的Receiver（如kafka、flume、socket）都会在两台机器上作数据备份。若是开启了write ahead logs的功能，建议把StorageLevel改为MEMORY_AND_DISK_SER。修改的方法是，在建立RDD时由参数传入。
(1)从high-level的角度来看，二者并无大的差异。都是将mapper(Spark中是ShuffleMapTask)的输出进行partition，不一样的partition送到不一样的reducer(Spark里的reducer多是下一个stage的ShuffleMapTask，也多是ResultTask)。Reducer之内存作缓冲区，边shuffle边aggregate数据，等数据aggregate好以后再进行reduce()(Spark里多是后续的一系列操做)
(2)从low-level的角度来看，二者差距不小。Hadoop MapReduce是sort-based，进入combiner()和reduce()的records必须先sort。这样的好处在于combiner()/reduce()能够处理大规模的数据，由于其输入数据能够经过外排获得(mapper对每段数据先作排序，reducer的shuffle对排好序的每段数据作归并)。目前spark选择的是hash-based，一般使用HashMap对shuffle来的数据进行aggregate，不会对数据进行提早排序。若是用户须要进行排序的数据，那么要本身调用相似SortByKey()的操做。
(3)从现实角度来看，二者也有不小差距。Hadoop MapReduce将处理流程划分出明显的几个阶段：map()，spill，merge，shuffle，sort，reduce()等。每一个阶段各司机制，能够按照过程式的编程思想来逐一实现每一个阶段的功能。在Spark中，没有这样功能明确的阶段，只有不一样的stage和一系列的transformation()，因此spill、sort、aggregate等操做须要蕴含在transformation()中。若是咱们将map()端划分数据、持久化数据的过程称为shuffle write，而将reducer读入数据、aggregate数据的过程称为shuffle read。那么在spark中，问题就变成怎么在job的逻辑或者物理执行图中加入shuffle write、shuffle read的处理逻辑，以及两个处理逻辑怎么高效实现。Shuffle write因为不要求数据有序，shuffle write的任务很简单：将数据partition好，并持久化。之因此要持久化，一方面是要减小内存存储空间压力，另外一方面也是为了fault-tolerance。
A list of partition
一个RDD有一系列的分区/分片
A function for computing each split/partition
对RDD的每个分区/分片都做用同一个函数
A list of dependencies on others RDDs
有一些依赖，在其余的RDD上
Optionally,a Partitioner for key-value RDDs(e.g to say that the RDD is hash-partitioned)
可选的，对于key-value的RDD的分区策略。
Optionally,a list of preferred locations to compute each split on(e.g. block locations for an HDFS file)
可选的，数据在哪儿优先把做业调度到数据所在节点进行计算：移动数据不如移动计算
优点：
1.速度快
2.其次，Spark是一个灵活的运算框架，适合作批次处理、工做流、交互式分析、流量处理等不一样类型的应用，所以spark也能够成为一个用途普遍的运算引擎，并在将来取代MapReduce的地位
3.最后，Spark能够与Hadoop生态系统的不少组件互相操做。Spark能够运行在新一代资源管理框架YARN上，它还能够读取已有并存放在Hadoop上的数据，这是个很是大的优点
劣势：
1.稳定性方面
2.不能处理大数据
3.不能支持复杂的SQL统计
Spark的shuffle整体而言就包括两个基本的过程：Shuffle write和Shuffle read。ShuffleMapTask的整个执行过程就是Shuffle write。将数据根据hash的结果，将各个Reduce分区的数据写到各自的磁盘中，写数据时不作排序操做。
首先是将map的输出结果送到对应的缓冲区bucket中，每一个bucket里的文件都会被写入本地磁盘文件ShuffleBlockFile中，造成一个FileSegment文件。
Shuffle Read指的是reducer对属于本身的FileSegment文件进行fetch操做，这里采用的netty框架，fetch操做会等到全部的Shuffle write过程结束后再进行，.reducer经过fetch获得的FileSegment先放在缓冲区softBuffer中，默认大小45MB。
1.构造Spark Application的运行环境(启动SparkContext)，SparkContext向资源管理器(能够是standalone、Mesos或Yarn)注册并申请运行Executor资源；
2.资源管理器分配Executor资源，Executor运行状况将随着心跳发送到资源管理器上；
3.SparkContext构建DAG图，将DAG图分解成Stage，并将Taskset发送给TaskSchedular。Executor向SparkContext申请Task，TaskSchedular将Task发送给Executor运行同时SparkContext将应用程序代码发送给Executor。
4.Task在Executor上运行，运行完毕释放全部资源。
1.下降批次处理时间：
①数据接收并行度。
(1)增长DStream：接收网络数据(如Kafka，flume，Socket等)时会对数据进行反序列化再存储在Spark，因为一个DStream只有Receiver对象，若是成为瓶颈可考虑增长DStream。
(2)设置”spark.streaming.blockInterval”参数：接受的数据被存储在Spark内存前，会被合并成block，而block数量决定了task数量；举例，当批次时间间隔为2秒且block时间间隔为200毫秒时，Task数量约为10；若是Task数量太低，则浪费了cpu资源；推荐的最小block时间间隔为50ms。
(3)显式对Input DStream从新分区：再进行更深层次处理前，先对输入数据进行从新分区。
②数据处理并行度：reduceByKey，reduceByKeyAndWindow等operation可经过设置”spark.default.parallelism”参数或显式设置并行度方法参数控制。
③数据序列化：可配置更高效的kryo序列化。
2.设置合理批次时间间隔：
①原则：处理数据的速度应大于或等于数据输入的速度，即批次处理时间大于或等于批次时间间隔。
②方法：
(1)先设置批次时间间隔为5~10秒数据输入速度；
(2)再经过查看log4j日志中的”Total delay”，逐步调整批次时间间隔，保证”Total delay”小于批次时间间隔。
3.内存调优：
①持久化级别：开启压缩，设置参数”spark.rdd.compress”；
②GC策略：在Driver和Executor上开启CMS(Content Management System 内容管理系统)
Yarn:你只须要一个节点，而后提交做业便可。这个是不须要spark集群的(不须要启动master和worker)
Standalone：你的spark集群上每一个节点上都要部署spark，而后须要启动spark集群。
Spark on Yarn支持client和cluster模式：driver运行在哪里
Client：driver运行在本地，提交做业的进程是不能中止的，不然做业就挂了。
Cluster：提交完做业，那么提交做业端就能够断开了，由于driver运行在am(application master)端。
[image:C7BD4379-4E3F-4E6A-8B48-A5B00A8C5A47-1000-0000244C808254F9/20181009222431411.png]
Spark中的内存管理主要分为两个方面：执行和存储。
执行端的内存主要是涉及到shuffle，join，sorts和aggregatations时的计算，存储端的内存主要涉及到cache。在spark中，执行和存储都是共享一个统一的region。当执行端没有使用内存时，存储端就能得到全部的内存信息，反之同样。在必要的时候，执行能够剔除存储，可是存储的时候能够设置一个阈值。
还能够看一个RDD消耗多少内存，在webUI或者使用SizeEstimator’s estimate方法。
内存使用的百分比是(堆内存-300MB)*0.6，执行和存储各占50%
使用广播变量在sparkContext中，能够大幅下降每个序列化task这个对象的大小，集群中启动一个job的成本也会下降。若是你的task中使用了一个大对象(large object)，考虑把他优化成一个广播变量。一般来讲，一个task大于20KB就值得优化。
数据本地性是有很大的影响在Spark job的程序中。若是数据和代码在一块儿，计算速度就会很是快。可是若是数据和代码是分开的，一个必需要移动到另一个上去。一般状况下是把序列化后的代码移动到数据所在的节点上，由于代码的大小比数据小不少(移动计算，而不是移动数据)。Spark构建的调度就是基于数据本地性。
数据本地性指的是数据和代码有多近(close)。由近及远有下面locality level：
1.PROCESS_LOCAL：数据在一个相同的正在运行的代码的JVM中。
2.NODE_LOCAL：数据在同一个节点。
3.NO_PREF：数据无论在哪里均可以快速的访问到。(无本地性)
4.RACK_LOCAL:数据在相同的机架上。可是数据在同一个机架的不一样server上，须要经过网络传输。
5.ANY：数据在网络的其余地方，不在一个机架上。
Spark会优先安排做业在最佳的locality level上，可是不太可能。
1)与其余计算框架共享集群资源（eg.Spark框架与MapReduce框架同时运行，若是不用Yarn进行资源分配，MapReduce分到的内存资源会不多，效率低下）；资源按需分配，进而提升集群资源利用等。
2)相较于Spark自带的Standalone模式，Yarn的资源分配更加细致
3)Application部署简化，例如Spark，Storm等多种框架的应用由客户端提交后，由Yarn负责资源的管理和调度，利用Container做为资源隔离的单位，以它为单位去使用内存,cpu等。
4)Yarn经过队列的方式，管理同时运行在Yarn集群中的多个服务，可根据不一样类型的应用程序负载状况，调整对应的资源使用量，实现资源弹性管理。
2种类型：1）result task类型，最后一个task，2）是shuffleMapTask类型，除了最后一个task都是。
rdd的mapPartitions是map的一个变种，它们均可进行分区的并行处理。
二者的主要区别是调用的粒度不同：map的输入变换函数是应用于RDD中每一个元素，而mapPartitions的输入函数是应用于每一个分区。
假设一个rdd有10个元素，分红3个分区。若是使用map方法，map中的输入函数会被调用10次；而使用mapPartitions方法的话，其输入函数会只会被调用3次，每一个分区调用1次。
这两个方法的另外一个区别是在大数据集状况下的资源初始化开销和批处理处理，若是在map和mapPartition中都要初始化一个耗时的资源，而后使用，好比数据库链接。在上面的例子中，mapPartition只需初始化3个资源（3个分区每一个1次），而map要初始化10次（10个元素每一个1次），显然在大数据集状况下（数据集中元素个数远大于分区数），mapPartitons的开销要小不少，也便于进行批处理操做。
mapPartitionsWithIndex和mapPartitons相似，只是其参数多了个分区索引号。
当SparkContext链接到集群管理器时，它会在集群中的节点上获取Executor。 executor是Spark进程，它运行计算并将数据存储在工做节点上。 SparkContext的最终任务被转移到executors以执行它们。