MapReduce
一、简介
MapReduce起源，在介绍大数据编年史时有提到Google最早在04年发表论文MapReduce，之后Doug Cutting基于这篇论文通过Java做了开源实现，Mapredce如今是作为Hadoop的核心组件之一，而HDFS是Hadoop的另外一个核心，此外还有Hadoop2.X之后推出的YARN。
关于MapReduce的学习，先来看一下他的核心设计思想：“分而治之，分布式计算”以及“移动计算，而非移动数据”。
“分而治之”，分治思想是MapReduce当中一个非常重要的思想。大数据场景当中我们处理的数据量非常之大，任务也很重，于是我们通常采用分治的思想，把这样一个大任务拆分为数个本质相同却又互相独立的小任务，这些小任务同时进行计算，后对计算结果进行汇总，这样子的话会比我们单进程的计算速度要快很多。
“移动计算，而非移动数据”，在早期大家作分析，当指定某一台服务器数据在计算时，首先会将数据从其他服务器通过网络传输到这一台，然后在这一台上边完成计算，这样子的设计在数据量比较小的情况下没有太大问题，但是当数据量非常大，假如现在有10TB数据要计算，将这些数据传输、“移动”到这一台服务器的话，它是否能承受呢？显然是不可以的，像我们现在讲到的这个场景里，计算过程当中我们移动的是数据，何为移动计算呢？就是既然现在数据已经按不同的block拆分之后分散的存储在不同的DataNode服务器上，可以让每一台数据存放结点（服务器）先将各自结点上的数据块计算完，然后再将最终结果做一个整理，也就是说数据在哪里、就在哪里计算，过程当中只是将计算任务移动到对应的数据结点来进行处理。这样的设计，一方面减少数据传输的IO成本，另一方面也高效利用了各个服务器的性能。
MapReduce有几个特点：
·移动计算而不移动数据：分布式计算，计算跟着数据走，数据存放在哪就在哪里进行计算，极大的减少了IO的开销。
·良好的扩展性：分布式计算框架拥有相当良好的扩展性，随着节点数量的增加，单个节点的计算量减小，整体的计算能力近乎线性的递增。
·高容错性：计算任务失败，会自动恢复计算，拥有良好的容错性。
·状态监控：提交任务之后，任务具体执行在哪个节点，具体执行到哪个阶段，在后台或者监控界面，我们都能实时的进行监控。
·适合于海量数据的离线批处理：MapReduce在设计之初的目标就是面向离线批处理，特别是大吞吐量的离线处理场景，更适合于MapReduce。
·降低了分布式编程的门槛：大部分操作MapReduce已经实现，我们仅仅需要在特定的部分编写我们自己的业务逻辑，极大的减少了工作量，同时也降低了编程的门槛。
适用场景：
对于高吞吐量要求的离线批处理场景，MapReduce都是适用的，如数据统计，搜索引擎构建索引，海量数据查询等。并且对于复杂数据分析算法的实现，MapReduce也是十分适用的。
相对的，MapReduce也有一些并不适用的场景：
对于实时计算，MapReduce是做不到的，MapReduce的设计之初是为了解决大吞吐量的场景，即某时刻数据计算量极大的场景，极大的数据量以及基于磁盘进行计算，就必然牺牲计算的速度，所以MapReduce任务的计算时间一般是分钟级别的。而对于秒级或者毫秒级别的敏捷计算，MapReduce并不适用。
其次是流计算，MapReduce同样不适用。首先流处理要求低延时，这一点MapReduce做不到，其次MapReduce要求输入的数据必须是静态的，而不能动态变化，所以对于数据实时发生变化的流式计算，MapReduce同样不适用。
最后是DAG计算，就是指多个作业间存在依赖关系，前一个的输出是后一个的输入，这种情况下虽然我们可以通过写多个MapReduce来进行实现，但是每一个MapReduce计算后的结果都要进行一次落盘，写入HDFS，所以从性能的考虑来说，MapReduce也并不太适合这种场景。
二、原理
通过词频统计（WordCount）这个例子来理解一下MapReduce的编程思想。
首先是作业（job）和任务（task）：
1. 作业（job）：作业是客户端提交请求执行的一个单元，它包括数据、计算框架以及一些配置信息等。
2. 任务（task）：是作业细分之后的细分工作单元，如MapReduce中的Map Task和Reduce Task。
其次是MapReduce。
MapReduce划分为四个阶段，分别为：Split、Map、Shuffle、Reduce。
1. Split阶段，主要负责“分”，这个阶段会由MapReduce自动将一个大文件切分成多个小的split文件片段，split只是逻辑概念，仅包含如数据起始位置，长度，所在位置等描述信息。2.x当中默认的切分规则，一个split刚好为一个block大小128M。那么10TB的数据文件，此时就会划分为像图中一样多个小split片段，每一个split交由一个Map Task处理。
2. Map阶段，会处理经过Split阶段切分好的数据片段，每一个split将对应一个Map的任务，也就说像图中所画Split切分为三个片段，分别对应着三个Map Task任务。Map阶段需要开发人员自己按照业务做实现，并且当我们分析的数据是HDFS当中文本数据时，他会一行一行来进行读取，最终需要按照Key-Value形式输出。那在WordCount案例中，读到每行数据时我们按照文本的分隔符将文本切分为一个一个单词，最后以单词作为key、1作为value进行输出。这样输出之后，最终对于每一个单词我们只要将1做累加就可以得到结果。
3. Shuffle阶段，他会完成分区、排序、分组等操作，分区决定了Map任务交由哪个Reduce任务处理，Reduce任务决定了有多少个分区，因为这个过程十分复杂，我们将在后边展开来介绍。先分析WordCount，默认Shuffle阶段会将Map阶段输出的Key-Value键值对按照单词的顺序做排序、分组，最终将相同的单词划分到一组，交给下个阶段Reduce来处理。
4. Reduce阶段，和Map一样都需要开发人员自己实现，它所处理的数据是Map输出之后经过Shuffle排好序、分好组的数据，那么在WordCount当中，Reduce任务每次处理的都是单词相同的一组数据，这段代码实现就很简单我只要对于这一组数据当中的Value进行累加，即可得到一个单词的数量，当Reduce所有任务执行完成即把每组单词数据处理完成之后，即可拿到最终的结果。
Job是客户端请求执行的一个工作单元，Task是一个Job的细分。
数据在输入时会被切分成等长的小数据块。默认Split大小等于Bock大小。如果数据存储在HDFS上，因为HDFS在数据存储时，相当于做了一次数据的切分，那么Mapreduce就不需要再做物理切分的操作。
Map阶段是做部分结果的合并，输入的是Split切片，对每个切片数据计算出中间结果（key-value格式）后输出。
Reduce阶段是对Map阶段部分结果的合并。输出的结果也是key-value的形式。
Shuffle阶段是Map和Reduce阶段的中间环节。需要将key值相同的Map中间结果分发到同一个Reduce节点上进行结果的汇总。
图中展示了Mapreduce的执行流程。首先数据要被Split切分，但是因为存储在HDFS上，所以数据已经被切分成了Block块，那接下来就会在每个Block块上分发一个Map作业进行中间结果的计算，计算结果保存为key-value的形式。此时shuffle阶段负责将Key值相同的数据分发到同一个Reduce节点上进行计算。Shuffle对Key值进行Hash取模，然后按照Reduce的个数形成对应的文件。Reduce节点会去Map节点去取自己的文件，取到之后进行合并。合并成大文件之后，在Reduce节点进行结果的汇总，最终结果保存到HDFS中。
MapReduce Shuffle阶段：
Shuffle连接了Map以及Reduce，它在Map以及Reduce两台服务器上都有执行。
在Map这端，Map做输出时会先将数据写入一个环形内存缓冲区，数据在写入缓冲区时会执行分区以及排序的方法。但是缓存区不会特别大（默认大小是100mb），所以设定有阀值，默认是0.80，同时Map输出时还会启动一个守护线程，当缓冲区的内存达到了阀值的80%时候，这个守护线程就会把内容溢写到磁盘上，这个过程就是图中的spill，另外的20%内存可以继续写入要写进磁盘的数据，写入磁盘和写入内存操作是互不干扰的。在这个过程中还会执行分区Partition、排序Sort的操作。分区Partition默认规则是按照Map输出Key的Hash值和Reduce数取模，排序Sort，默认是按照Map输出的Key的ASCII码值（字典序）进行排序。数据由环形缓冲区溢写之后的小文件内部都是分好区、排好序的效果，即每个分区内部有序。Map源源不断地输出，不断进行溢写，最终在Map这端会有一堆小文件，这些小文件需要做一个合并，合并成一个大文件，大文件内部是分区有序的。
在Reduce这端，Reduce会根据Map那端分区之后的结果，主动将对应分区数据拉取过来进行处理，这时候拉取过来和Map那端类似，数据首先写入到内存缓存区内，快写满、到达溢写阈值之后会发生溢写到磁盘，数据不断拉取过来不断作溢写操作，最终也会生成一系列小文件，小文件最终也会合并成大文件，大文件最后交给Reduce任务去处理。在这个过程中，数据在Reduce端还会再做一次排序，一般称为Group分组，他排序之后会把数据划分为一组一组的效果，ReduceTask任务在处理数据时，每一次处理一组数据。默认的Group规则，也是按照Key的ASCII码值进行排序（字典排序）。
Shuffle详解部分查看PPT中的详细描述，或结合上一页PPT内容进行理解。
2.x中出现了新的组件分布式通用资源管理框架YARN，所以Hadoop2.x之后作业调度基本由YARN来完成。在YARN中，不同于Hadoop1.x的JobTracker统一管理资源以及任务，它是由一个全局的资源管理器ResourceManager以及每个作业一个的任务管理器ApplicationMaster分别管理资源和任务。并且YARN中将资源抽象化为Container，任务就运行在这些NodeManager上的Container中，NodeManager负责管理单节点的资源。其中NodeManager和ResourceManager定期保持Hearbeat通信，来进行资源管理，ApplicationMaster由ResourceManager负责创建，作为一个进程运行在一个NodeManager上，来负责某个作业的任务管理。客户端将Job提交给ResourceManager，ResourceManager首先会创建一个ApplicationMaster，后由ApplicationMaster将Job拆分为多个Map Task以及Reduce Task，并向ResourceManager申请资源，之后ApplicationMaster拿到申请的资源，并在相应的NodeManager上运行这些Task，如果有任务运行失败，同样是由ApplicationMaster申请新的资源，并重新运行这些失败的任务，直到所有Task运行完成。
三、作业管理
可以使用hadoop jar将作业提交到hadoop中，然后通过yarn application -list命令查看提交的作业，可以使用yarn application -kill去终止这个作业。
使用hadoop jar命令将任务提交到集群中运行。
如果作业在运行中异常报错，需要排查yarn的日志去分析错误原因。
Spark
一、简介
Spark最初在2009年由加州大学伯克利分校的AMP实验室开发，并于2010年成为Apache的开源项目之一，如今正在逐渐替代MapReduce成为Hadoop平台之上非常重要的计算框架。值得注意的是，Spark并不是一个单一的计算框架，它集成了适合批处理的Spark Core、结构化数据处理的Spark SQL、实时流处理的Spark Streaming、机器学习组件Spark MLlib以及图计算的Spark GraphX多种组件，它不止在计算速度上有了质的飞跃，更加统一多种功能，做到了通用性的实现。
Spark的提出主要解决解MapReduce实现的一些弱点：难以支持复杂应用场景，如机器学习、流式计算、图计算等；迭代式计算的效率低下等问题。
Spark基本特点：
1、 基于内存计算，对比MapReduce过程当中非常多的内存磁盘数据交互、性能比较低，而Spark计算全部在内存当中完成，不同结点直接数据传输全部通过网络完成，所以速度上比MapReduce更加高效；
2、 基于DAG优化任务流程，支持迭代式计算，利用自身的DAG引擎，减少中间计算结果写入HDFS的开销。
3、 利用自身的多线程池模型，极大的减少了任务启动时的开销，避免了很多的排序以及IO操作。
4、 多种组件支持批处理，流处理，交互式计算，机器学习等多种开发场景，真正做到通用易用。
5、 丰富的API支持，如Scala、Java、Python等，一般开发过程中建议使用Scala（Spark源码为Scala、且Scala效率较于Java会更好）
基于这些优势，Spark比MapReduce更加适用于以下场景：机器学习和图应用中常用的迭代算法（每一步对数据执行相似的函数）以及交互式数据挖掘工具。
二、原理
Spark部分，我们重点先理解RDD的概念。首先来看RDD的内容：
RDD（Resilient Distributed Datesets、弹性分布式数据集）是Spark特有的数据模型，Spark当中的计算都是通过操作RDD来完成的。DAG（有向无环图），RDD各项操作之间的相互依赖会被转成DAG，DAG划分不同的stage阶段，由不同的task任务运行。
RDD同时也是Spark的基本计算单元，是对分布式内存的抽象使用，实现了以操作本地集合的方式来操作分布式数据集的抽象实现。RDD是Spark最核心的东西，它表示已被分区，不可变的并能够被并行操作的数据集合，不同的数据集格式对应不同的RDD实现。RDD必须是可序列化的。RDD可以cache到内存中，每次对RDD数据集的操作之后的结果，都可以存放到内存中，下一个操作可以直接从内存中输入，省去了MapReduce大量的磁盘IO操作。对应用透明，开发人员只需要对于RDD进行操作即可不需要其他的处理。RDD的创建操作只能是基于稳定的数据集或者已有的RDD，整个Job任务在计算过程中如果出现错误，可以通过这一系列的转换、算子追述到之前的操作，自动重构从而保证计算的正确性。
RDD的操作分为转换和动作。
由一个RDD处理生成另一个RDD，这个操作称之为转换transformation。转换transformation是RDD的一种算子，他记录了对于数据集的逻辑操作，Spark当中有很多的算子或者函数的支持，例如：map、filter、union、distinct等等，像这一类操作仅仅只是转换逻辑的上的操作，并不需要将数据进行物化、持久化。
Action的触发是会真正执行计算、执行操作的，例如count做一次聚合；而对比map仅仅只是一个逻辑的概念，当他遇到count的时候才会开始触发执行。
RDD之间相互的这些操作，一次动作之后到另一个动作执行，这种关系称之为依赖，Spark当中把依赖划分为宽依赖以及窄依赖：
窄依赖是指 父 RDD 的每个分区都只被子 RDD 的一个分区所使用。相应的，那么宽依赖就是指子RDD的分区依赖于父RDD的所有分区。例如图中，Map就是一种窄依赖，而Join 则会导致宽依赖。
transformations和actions的本质区别在于：transformations是从一个RDD到另一个RDD；而actions是从一个RDD到最终结果。Spark计算核心就是对RDD的这两种操作，如果任务定义的lineage链非常长恢复会比较耗费时间，因为恢复需要追溯到之前的节点重新来过，并且对于宽依赖来说，如果出错则需要把父RDD分区的数据重新计算一次才可以完成。所以对于这些问题，实现过程中，可以设置检查点来解决，或者将中间的RDD数据物化的磁盘当中。
接下来来看一个例子，通过这个例子，以WordCount为例学习Spark程序的执行流程。首先获得HDFS中的某一个数据集，并把它封装成一个RDD。之后，对这个RDD执行了flatMap来对数据进行分割压平，Map把每个分割后的数据转换成键值对的形式，并对每个键值对的Value赋值为1，最后对这些相同Key的键值对进行整合累加，来完成之前已经用MapReduce实现过一次的WordCount。值得注意的是对数据集的操作是以对RDD操作的形式完成的，flatMap和map都是Transformation算子，并没有真正触发执行，仅仅只是逻辑操作而已，而之后的reduceByKey算子，统计数据集中元素个数，此时才会真正执行操作。
在程序运行的时候，首先会运行Driver，Driver相当于是整个任务的管理程序，负责对任务进行解析、分发、监控等。Driver中包含的SparkContext是Spark的运行环境。
Driver运行起来之后，会向ClusterManager主节点去申请资源，申请到的资源是在WorkerNode上封装好的Executor容器，容器包含了程序运行的CPU、内存等资源。
然后Driver会将Task分发到这些Executor中进行执行，执行过程中，会时刻监控这些Task的运行情况，并做实时的管理和调度。
·Driver：每一个Spark程序都会自带一个Driver驱动器，程序的main函数运行在Driver中。Driver主要负责Spark程序的解析，划分Stage，以及分发各个任务到Executor中去执行，并且Driver也会创建一个SparkContext。
·SparkContext：SparkContext负责初始化运行环境，加载配置信息，以及创建DAGSchedule和TeskSchedule。
·Executor：执行器Executor负责执行Driver分发的多个任务，一个节点可以启动多个Executor，每个Executor通过多线程运行多个任务。
·Task：Task是Spark运行的基本单位，一个Task为一个Executor线程，负责处理一个RDD分区的计算逻辑。
Spark支持多种运行模式，首先Local模式，一般用于单机执行，通常用于测试环境。
Standalone模式是将Spark程序运行在Spark的独立集群中。这种情况下Driver在WorkNode节点运行，然后Task被分发到Worker节点的Executor容器中运行。
Yarn模式是比较常见的一种模式，Spark将任务提交到yarn上去运行。这种模式根据Driver的位置不同，又细分为Client和Cluster两种模式。Cluster模式是生产中常见的运行模式，这种模式是由ApplicationMaster来进行资源的申请，然后作业的解析和管控由Driver来执行，此时ApplicationMaster和Driver是在一起的，并运行在NodeManager上。而Client模式则是Driver运行在了Client端，Application运行在NodeManager上负责资源的申请，这样的好处是Driver可以实时的将作业的信息显示在控制台上，便于作业的监控和调优，常用于交互和调试。
当我们提交了Spark应用程序之后，首先要在Driver中把我们的Spark程序转化成对应的逻辑查询计划，之后根据生成的逻辑查询计划，生成计算机可真正识别的物理查询计划，此时，相当于有了我们细分的各个具体任务，最后再把这些任务进行分发调度。Executor拿到我们分发的这些个任务，开始任务的真正执行。
逻辑查询计划的生成基本上就是我们所写的计算逻辑，根据RDD之间的流程关系等，生成对应的逻辑查询计划。
物理查询计划的生成，依赖于我们的逻辑查询计划。首先根据我们RDD的种类以及对应的宽窄依赖关系，生成多个Stage，每个Stage之间也会有对应的逻辑关系，如图所示。最后由我们的多个Stage，组成了我们最后的DAG。
当拿到了多个Stage，提交给Driver来执行的时候，基本就是这个样子。以图为例，首先Stage1中全部是一些Transformation操作，而Stage1到Stage2之间出现了宽依赖关系，也就是出现了Action操作。这些个动作转换，就是要提交给Executor来执行的Task任务，所有Task任务的分配以及监控，都是由SparkContext来完成的。
DAG的含义：一个有向图无法从任意顶点出发经过若干条边回到该点。
RDD操作中的一系列依赖关系，Spark后期会转换为DAG来进行表示。
如图所示，左边为RDD创建、操作，rdd1和rdd2做Join之后分组、过滤之后进行count统计，中间一部分是Spark的DAGScheduler计算RDD之间的依赖关系、生成相应的DAG，DAGScheduler对这些依赖关系形成的DAG进行Stage划分，划分的规则很简单，从后往前回溯，遇到窄依赖加入本stage，遇见宽依赖进行Stage切分。完成了Stage的划分，DAGScheduler基于每个Stage生成TaskSet，并将TaskSet提交给TaskScheduler。TaskScheduler 负责具体的task调度，在Worker节点上启动task。最右边这些不同的stage对应不同的task任务进行执行。
划分Stage的一个主要依据是当前计算因子的输入是否是确定的，如果是则将其分在同一个Stage，避免多个Stage之间的消息传递开销。如图所示是stage的一个划分，图中被划分为三个stage。
三、任务监管
Spark提供相应任务监控UI，可通过以下地址访问：http://{inceptor_server_ip}:4040
如图所示，我们可以通过访问TDH中Inceptor Server所在节点的4040端口，来监控管理Spark任务，在界面上我们可以很方便的做到Spark任务的监控管理。
