一、Hadoop架构：

1、它是一个主从结构，主节点被称为master，从节点被称为slave。

2、它是一个分布式架构，我所理解的分布式就是分组合并的，分组是指比如一个较大数据，一台电脑处理不了，然后我们将这个数据分成多份，每份存储在从属主机，并且在从属主机上面进行计算；合并是指，将从属主机的计算结果合并成一个。

3、Hadoop主要包括HDFS和MapReduce。

4、相较于Hadoop1.0，Hadoop2.0以后引入了yarn。HDFS是一个分布式存储系统。MapReduce是分布式的计算框架。yarn是一个资源调度的工具。Hadoop2.0加入了zk的支持实现比较可靠的高可用。

二、Hadoop中的HDFS：

1、HDFS的启动过程：（1）加载元文件。（2）加载日志文件。（3）设置检查点。（4）进入安全模式。作用是：检查数据块的副本率，冗余度是否达到要求。

2、HDFS中有三个"士兵"：（1）NameNode。（2）、DataNode。（3）Secondary NameNode。

3、NameNode的职责：

（1）、加载fsimage和edists文件到内存中。

（2）、接受客户端发起的增删改查请求。

（3）、记录操作日志，更新滚动日志。

（4）、在内存中对数据进行增删改查。

4、DataNode的职责：

（1）、保存数据块，默认是128M，Hadoop1.0默认是64M。

（2）、存储在磁盘上，包括两个文件，一个数据本身，一个是元数据包括数据块长度，块数据校验和以及时间戳。

（3）、DataNode启动后向NameNode注册，通过后，周期性（64分钟）的向NameNode上报所有的信息。

（4）、心跳是三秒一次，心跳返回结果包括NameNode给该DataNode的命令。如果十分钟都没有收到该DataNode的心跳，则该节点不可用。

5、Secondary NameNode：

（1）、向NameNode询问是否需要checkPoint。

ps：检查点触发的条件是：①时间快到了，60分钟。②日志文件edits中数据满了，64M。

（2）、请求执行checkPoint。

（3）、拷贝edits文件和fsimag文件到SecondaryNameNode中。

（4）、将拷贝过来的edits文件和fsimag文件进行合并。

（5）、生成新的fsimage命名为fsimage.chkpoint。

（6）、将fsimage.chkpoint拷贝到NameNode中去。

（7）、重新命名生成fsimage。

6、HDFS上传文件的流程：

（1）、客户端发出文件上传的请求。

（2）、客户端创建出DistrbutedSystem。

（3）、DistributedSystem和NameNode建立RPC通信，得到一个NameNode的代理对象。

（4）、请求创建文件的元信息。

（5）、NameNode创建出文件的元信息。元信息包括文件名，文件的大小，路径，数据块数量，数据块分布在DataNode的信息。

（6）、NameNode缓存元信息。

（7）、如果满了就采用LRU置换算法，生成fsimage文件。

（8）、返回给客户端文件的元信息。

（9）、创建输出流。

（10）、上传第一个数据块到DataNode上面去。

（11）、DataNode之间根据冗余信息，进行复制。

（12）、FSDataOutputStream循环上传数据。

7、HDFS下载文件的流程：

（1）、客户端发出文件下载的请求。

（2）、客户端创建出DistrbutedSystem。

（3）、DistributedSystem和NameNode建立RPC通信，得到一个NameNode的代理对象。

（4）、得到文件的元信息。

（5）、NameNode在内存中查找文件的元信息。

（6）、在fsimage中查找。

（7）、返回给客户端文件的元信息。

（8）、创建一个输入流。

（9）、下载第一个数据块，多线程下载数据块。

三、HDFS和MapReduce的联系：

1、针对HDFS而言：NameNode、DataNode、Secondary NameNode。

2、针对MapReduce而言：JobClient、JobTracker、TaskTracker。

ps：接受端通过JobClient类将已经配置参数打包成jar文件的应用存储到hdfs，并把路径提交到Jobtracker,然后由JobTracker创建每一个Task（即 MapTask 和 ReduceTask）并将它们分发到各个TaskTracker服务中去执行。

3、针对Hadoop而言：NameNode和JobTracker属于Master，DataNode和JobTaskTracker属于Slave。

4、JobTracker的Recover机制：如果设置了JobTracker Recover机制，那么在JobTracker重启的时候会触发Recover，在重启之前，histroy logs中会记录各个作业的运行状态，待JobTracker重启的时候，JobTracker会自动提交这些作业，并只对未完成的task进行重新调度。

四、MapReduce中的MapTask：

1、MapTask工作流程：Read阶段-->Map阶段-->Collect阶段-->Spill溢写阶段-->Combiner阶段

2、各个阶段：

（1）、Read阶段：maptask调用InputFormat，InputFormat又调用RecordReader从输入文件中解析出一个个K/V。

（2）、Map阶段：该节点主要是将解析出的key/value交给用户编写map()函数处理，并产生一系列新的key/value。

（3）、Collect收集阶段（partition阶段）：在用户编写map()函数中，当数据处理完成后，一般会调用OutputCollector.collect()输出结果。在该函数内部，它会将生成的key/value分区（调用Partitioner），并写入一个环形内存缓冲区中。

（4）、Spill溢写阶段：当环形缓冲区达到80%时（环形缓冲区默认是100M），会将数据写到本地磁盘上生成一个临时文件。ps：将数据写入本地磁盘之前，先要对数据进行一次本地排序，并在必要时对数据进行合并、压缩等操作。（步骤1：利用快速排序算法对缓存区内的数据进行排序，排序方式是，先按照分区编号partition进行排序，然后按照key进行排序。这样，经过排序后，数据以分区为单位聚集在一起，且同一分区内所有数据按照key有序。

步骤2：按照分区编号由小到大依次将每个分区中的数据写入任务工作目录下的临时文件output/spillN.out（N表示当前溢写次数）中。如果用户设置了Combiner，则写入文件之前，对每个分区中的数据进行一次聚集操作。

步骤3：将分区数据的元信息写到内存索引数据结构SpillRecord中，其中每个分区的元信息包括在临时文件中的偏移量、压缩前数据大小和压缩后数据大小。如果当前内存索引大小超过1MB，则将内存索引写到文件output/spillN.out.index中。）

（5）、Combine阶段：当所有数据处理完成后，MapTask对所有临时文件进行一次合并，以确保最终只会生成一个数据文件，可避免同时打开大量文件和同时读取大量小文件产生的随机读取带来的开销。（步骤1：在进行文件合并过程中，MapTask以分区为单位进行合并。对于某个分区，它将采用多轮递归合并的方式。每轮合并io.sort.factor（默认10）个文件，并将产生的文件重新加入待合并列表中，对文件排序后，重复以上过程，直到最终得到一个大文件。

步骤2：当所有数据处理完后，MapTask会将所有临时文件合并成一个大文件，并保存到文件output/file.out中，同时生成相应的索引文件output/file.out.index。）

3、小文件优化：

4、MapTask的个数由什么决定：

（1）、一个job的map阶段MapTask个数，由客户端提交的job的切片个数决定。

（2）、每一个切片分配一个MapTask。

（3）、切片大小默认=blocksize。

（4）、切片时针对每一个文件单独切片，不考虑数据集整体。

五、MapReduce中的ReduceTask：

1、ReduceTask的工作机制：Copy阶段-->Merge阶段-->Sort阶段-->Reduce阶段

2、各个阶段：

（1）、Copy阶段：ReduceTask从各个MapTask上远程拷贝一片数据，并针对某一片数据，如果其大小超过一定阈值，则写到磁盘上，否则直接放到内存中。

（2）、Merge阶段：在远程拷贝数据的同时，ReduceTask启动了两个后台线程对内存和磁盘上的文件进行合并，以防止内存使用过多或磁盘上文件过多。

（3）、Sort阶段：按照MapReduce语义，用户编写reduce()函数输入数据是按key进行聚集的一组数据。为了将key相同的数据聚在一起，Hadoop采用了基于排序的策略。由于各个MapTask已经实现对自己的处理结果进行了局部排序，因此，ReduceTask只需对所有数据进行一次归并排序即可。

（4）、Reduce阶段：reduce()函数将计算结果写到HDFS上。

3、ReduceTask个数由什么决定：一个job的reduceTask数量是通过job.setNumReduceTasks(x)来设置的;x 为reduce的数量。不设置的话默认为 1。

六、MapReduce中的shuffle：

1、shuffle图解：

2、map端的shuffle：

（1）、partition：

（1）过程：经过map函数处理后输出新的<key,value>，它首先被存储到环形缓冲区的kvbuffer，环形缓冲区默认是100M，并且对每个key/value对hash一个partition值，相同的partition值为同一个分区。

（2）作用：分区之后，每个reduce就会处理对应的partition，减少reduce的压力。

（2）、sort/combiner/compress：

（1）过程：对环形缓冲区内的partition值和key值进行排序；如果用户设置了combiner，会对每个partition中的数据进行预处理，相当于是map端的reduce；如果用户设置了compress，会对combiner的数据进行压缩。

（2）作用：sort作用是在内部排序，减少reduce的压力；combiner作用是节省网络带宽和本地磁盘的IO；compress作用是减少本地磁盘的读写和减少reduce拷贝map端数据时的网络带宽。

（3）、spill：

（1）过程：因为环形缓冲区的内存不够用，所以必须要写到本地磁盘中。将排序好的数据spill到本地磁盘中。

（2）作用：数据量非常大，全部放到内存不现实，所以最后还是会存到本地磁盘中。

（4）、merge：

（1）过程：因为会产生多次spill，本身存放数据的out文件和存放数据偏移量索引index文件会产生多个，把多个文件合并在一起。

（2）作用：方便reduce的一次性拷贝。

3、reduce端的shuffle：

（1）、copy：

（1）过程：reduce拷贝map最终的输出的磁盘数据，一个reduce拷贝每个map节点的相同partition数据。

（2）作用：拷贝后的数据不止一份，先进行合并操作，为后面的排序做准备。

（2）、merge：

（1）过程：reduce拷贝map最终的输出的磁盘数据，一个reduce拷贝每个map节点的相同partition数据。

（2）作用：拷贝后的数据不止一份，先进行合并操作，为后面的排序做准备。

（3）、sort：这里和map端的一样，但是reduce端的缓冲区更加灵活一点，如果内存够用，就是内存到内存的merge，不够用了就是内存到磁盘的merge，最后是磁盘到磁盘的merge。

（4）、group：将排序好的数据进行分组，分组默认是将相同的key的value放在一起。作用是为了reduce函数更好的计算相同key值出现的次数。

4、shuffle的优化：

（1）、Map端的优化：（1）kvbuffer，默认是100M，可以通过参数mapreduce.task.io.sort.mb来修改，一般不修改。

（2）缓冲区阈值，一般是80%，可以通过mapreduce.map.sort.spill.percent来修改。

（3）合并spill文，mapreduce.task.io.sort.factor属性配置每次最多合并多少个文件，默认为10,即一次最多合并10个spill文件.如果spill文件数量大于mapreduce.map.combiner.minspills配置的数，则在合并文件写入之前，会再次运行combiner。如果spill文件数量太少，运行combiner的收益可能小于调用的代价。

（4）、对map输出进行压缩，在数据量大的时候，可以对map输出进行压缩,要启用压缩，将mapreduce.map.output.compress设为true，并使用mapreduce.map.output.compress.codec设置使用的压缩算法。

（2）、Reduce端优化：

（1）copy线程数量。copy是用来从map任务中提取数据的,默认为5个copy线程,可以通过mapreduce.reduce.shuffle.parallelcopies配置。

（2）内存分配：如果能够让所有数据都保存在内存中，可以达到最佳的性能。通常情况下，内存都保留给reduce函数，但是如果reduce函数对内存需求不是很高，将mapreduce.reduce.merge.inmem.threshold（触发合并的map输出文件数）设为0，mapreduce.reduce.input.buffer.percent（用于保存map输出文件的堆内存比例）设为1.0。

七、MapReduce中的高级特性：

1、MapReduce的高级特性：

（1）、序列化。（2）、排序。（3）、分区。（4）、合并。

2、序列化：

（1）、核心接口是：Writable，如果一个类实现了Writable接口，该类的对象可以作为key和value。

（2）、Java中的序列化是实现Serializable。

（3）、序列化的注意事项：1）、反序列化时，需要调用空参构造函数，所以必须有空参构造。2）、重写序列化方法。3）、重写反序列化方法（顺序必须和序列化的顺序一样）。4）、想要把结果显式在文件中，需要重写toString方法。5）、如果要自定义bean放在key中传输，则需要实现Comparable接口。

3、排序：

（1）、MapReduce自带的排序，默认是按照key排序。

（2）、自定义排序：需要实现writablecomparable接口，重写compareTo。

（3）、二次排序：一般使用SortComparator和GroupComparator，先是调用SortComparator，如果满足则调用GroupComparator函数，只要两个key相同，就会把它们的value放到一个value迭代器中，如果没有满足则调用key自己实现的compareTo方法。

4、分区：

（1）、分区是根据MR的输出<key，value>进行分区的。默认情况下，MR的输出只有一个分区，一个分区就是一个文件。

（2）、自定义分区：继承Partitioner，重写getPartition这个方法。

（3）、如果没有定义partitioner，那数据在被送达Reducer前是按照每一条数据的key的hashcode进行分区的。

5、合并（Combiner）：

（1）、Combiner是一种特殊的Reducer，它是对每一个mapTask的输出进行局部汇总，以减少网络传输量。

（2）、好处是：合并在Mapper端执行一次合并，用于减少Mapper输出到Reducer的数量，可以调高效率。

（3）、谨慎使用Combiner，Combiner的输出的k，v应该和Reducer的输入的k，v相对应。不能用的Combiner的例子，例如：求平均值。

（4）、Combiner和Reducer的区别在于运行的位置：Combiner是在每一个mapTask所在的节点运行。Reducer是接收全局所有Mapper的输出结果。

八、MapReduce中的调优：

1、map端的shuffle优化：

（1）kvbuffer，默认是100M，可以通过参数mapreduce.task.io.sort.mb来修改，一般不修改。

（2）缓冲区阈值，一般是80%，可以通过mapreduce.map.sort.spill.percent来修改。

（3）合并spill文，mapreduce.task.io.sort.factor属性配置每次最多合并多少个文件，默认为10,即一次最多合并10个spill文件.如果spill文件数量大于mapreduce.map.combiner.minspills配置的数，则在合并文件写入之前，会再次运行combiner。如果spill文件数量太少，运行combiner的收益可能小于调用的代价。

（4）、对map输出进行压缩，在数据量大的时候，可以对map输出进行压缩,要启用压缩，将mapreduce.map.output.compress设为true，并使用mapreduce.map.output.compress.codec设置使用的压缩算法。

2、reduce端的shuffle优化：

（1）、copy线程数量。copy是用来从map任务中提取数据的,默认为5个copy线程,可以通是mapreduce.reduce.shuffle.parallelcopies配置。

（2）、内存分配：如果能够让所有数据都保存在内存中，可以达到最佳的性能。通常情况下，内存都保留给reduce函数，但是如果reduce函数对内存需求不是很高，将mapreduce.reduce.merge.inmem.threshold（触发合并的map输出文件数）设为0，mapreduce.reduce.input.buffer.percent（用于保存map输出文件的堆内存比例）设为1.0。

3、资源相关参数：

1）、mapreduce.map.memory.mb 一个Map Task可使用的资源上限（单位:MB），默认为1024。如果Map Task实际使用的资源量超过该值，则会被强制杀死。

2）、mapreduce.reduce.memory.mb 一个Reduce Task可使用的资源上限（单位:MB），默认为1024。如果Reduce Task实际使用的资源量超过该值，则会被强制杀死。

3）、mapreduce.map.java.opts Map Task的JVM参数，你可以在此配置默认的java heap size等参数, e.g."-Xmx1024m -verbose:gc -Xloggc:/tmp/@taskid@.gc" （@taskid@会被Hadoop框架自动换为相应的taskid）, 默认值: ""4）、mapreduce.reduce.java.opts Reduce Task的JVM参数，你可以在此配置默认的java heap size等参数, e.g."-Xmx1024m -verbose:gc -Xloggc:/tmp/@taskid@.gc", 默认值: ""5）、mapreduce.map.cpu.vcores 每个Map task可使用的最多cpu core数目, 默认值: 16）、mapreduce.reduce.cpu.vcores 每个Reduce task可使用的最多cpu core数目, 默认值: 1

7）、yarn.scheduler.maximum-allocation-mb 8192 给应用程序container分配的最大内存

8）、yarn.scheduler.minimum-allocation-mb 1024 给应用程序container分配的最小内存

4、容错相关参数：

1）、mapreduce.map.maxattempts 每个Map Task最大重试次数，一旦重试参数超过该值，则认为Map Task运行失败，默认值：4。

2）、mapreduce.reduce.maxattempts 每个Reduce Task最大重试次数，一旦重试参数超过该值，则认为Map Task运行失败，默认值：4。

3）、mapreduce.map.failures.maxpercent 当失败的Map Task失败比例超过该值为，整个作业则失败，默认值为0. 如果你的应用程序允许丢弃部分输入数据，则该该值设为一个大于0的值，比如5，表示如果有低于5%的Map Task失败（如果一个Map Task重试次数超过mapreduce.map.maxattempts，则认为这个Map Task失败，其对应的输入数据将不会产生任何结果），整个作业扔认为成功。

4）、mapreduce.reduce.failures.maxpercent 当失败的Reduce Task失败比例超过该值为，整个作业则失败，默认值为0。5）、mapreduce.task.timeout Task超时时间，经常需要设置的一个参数，该参数表达的意思为：如果一个task在一定时间内没有任何进入，即不会读取新的数据，也没有输出数据，则认为该task处于block状态，可能是卡住了，也许永远会卡主，为了防止因为用户程序永远block住不退出，则强制设置了一个该超时时间（单位毫秒），默认是300000。如果你的程序对每条输入数据的处理时间过长（比如会访问数据库，通过网络拉取数据等），建议将该参数调大，该参数过小常出现的错误提示是“AttemptID:attempt_14267829456721_123456_m_000224_0 Timed out after 300 secsContainer killed by the ApplicationMaster.”。

5、数据倾斜：

（1）、数据倾斜的解决：（1）、抽样和范围分区———可以通过对原始数据进行抽样得到的结果集来预设分区边界值。（2）、自定义分区。（3）、Combine。

（2）、数据倾斜问题：①数据倾斜频率：某一个区域的数量要远远大于其他区域。②数据大小倾斜：部分记录的大小远远大于平均值。

九、思维导图：