标签：总结 切片 文件 对象 Hadoop 学习 MR 数据

Hadoop阶段学习总结

第一部分：HDFS相关问题

一、描述一下HDFS的数据写入流程

​ 首先由客户端想NameNode服务发起写数据请求，NameNode接收到请求后会进行基本验证，验证内容包括对请求上传的路径进行合法验证其次还要对请求的用户进行权限验证。验证没有问题后，NameNode会响应客户端允许上传。接下来客户端会对文件按照blocksize大小进行切块，切完块后依次以块为单位进行上传。此时客户端会请求上传第一个块信息，服务端接收到上传请求后会依据HDFS默认的机架感知原理默认情况下返回三台存放数据块副本的DataNode机器。客户端接收到机器列表后会依据网络拓扑的原理找到其中一台机器进行传输通道的建立，然后依次和三台机器进行串行连接，这样的连接方式主要的为了减轻客户端本地的IO的压力。当通道建立成功后，客户端会通过HDFS的FSOutputStream流对象进行数据传输，数据传输的最小单位为packet。传输过程中每太DataNode服务器是串行连接，依次将数据传递。最后一个数据块被传输完成后相当于一次写入结束，如果还有数据块需要传输，那就接着传输第二个数据块。

二、描述一下HDFS的数据读取流程

​ 首先和写数据一样，由客户端向NameNode发出请求，NameNode接收到请求后会进行文件下载路径的合法性校验以及权限验证。如果验证没有问题，就会给客户端返回目标文件的元数据信息，信息中包含目标文件数据块对应的DataNode的位置信息。然后客户端根据具体的DataNode位置信息结合就近原则网络拓扑原理找到离自己最近的一台服务器对数据进行访问下载，最后通过HDFS提供的FSInputStream对象将数据读取到本地。如果有多个块信息 就会请求多次DataNode直到目标文件的全部数据被下载。

三、简述HDFS的架构，其中每个服务的作用

​ HDFS是Hadoop架构中的负责完成数据的分布式存储管理的文件系统。非高可用HDFS集群工作的时候会启动三个服务，分别是NameNode 和 DataNode以及SecondaryNameNode 。其中NameNode是HDFS的中心服务，主要维护管理文件系统中的文件的元数据信息，DataNode主要负责存储文件的真实数据块信息，当然在DataNode的数据块信息中也包含一下关于当前数据块的元数据信息 例如 检验值 数据长度 时间戳等。在非高可用HDFS集群中 NameNode和DataNode可以理解为是一对多的关系。二者在集群工作中也要保持通信，通常默认3秒钟会检测一下心跳。最后SecondaryNameNode的工作很单一，就是为了给NameNode的元数据印象文件和编辑日志进行合并，并自己也保留一份元数据信息 以防NameNode元数据丢失后有恢复的保障。

四、HDFS中如何实现元数据的维护？

​ NameNode的元数据信息是通过fsimage文件 + edits编辑日志来维护的，当NameNode启动的时候fsimage文件和edits编辑日志的内容会被加载到内存中进行合并形成最新的元数据信息，当我们对元数据进行操作的时候，考虑到直接修改文件的低效性，从而不会直接修改fsimage文件，而是会往edits编辑日志文件中追加操作记录。当满足一定条件的时候 我们会让SecondaryNameNode来完成fsimage文件和edits 编辑日志文件的合并，SecondaryNameNode首先会让NameNode停止对正在使用的edits编辑日志文件的使用，并重新生成一个新的edits编辑日志文件。接着把NameNode 的fsimage文件和已停止的edits文件拷贝到本地在内存中将edits编辑日志文件的操作记录合并到fsimage 文件中形成一个最新的fsimage文件，最后会将这个最新的fsimage文件推送给NameNode并自己也备份一份。

五、描述一下NN和DN的关系，以及DN的工作流程

​ NameNode和DataNode从数据结构上看，就是一对多的关系，一个HDFS集群中是能只能有一个NameNode用于维护元数据信息，同时会有多个DataNode用于存储真实的数据块。当HDFS集群启动的时候，首先会进入到安全模式下，在安全模式下我们只能对数据进行读取不能进行任何写操作，此时集群的每一台DataNode服务器会向NameNode注册自己，注册成功后DataNode会上报自己的数据块详细信息，当数据块汇报满足 最小副本条件后，安全模式就自动退出。此后 DataNode和NameNode每三秒会通信一次，如果NameNode检测到DataNode没有响应，会继续检测 一直到10分30秒后还没有检测到 就确定当前DataNode不可用。

第二部分：MapReduce相关问题

一、描述一下手写MR的大概流程和规范

​ 首先，从MapReduce程序的结构划分可以分为三部分，第一是 程序的执行入口通常简称为驱动类，驱动类主要编写MR作业的提交流程以及自定义的一些配置项。第二是 Map阶段的核心类需要自定并且继承Hadoop提供的Mapper类，重写Mapper类中的map方法，在map方法中遍写自己的业务逻辑代码将数据处理后利用context 上下文对象的写出落盘。第三是 Reduce阶段的核心类同时也需要继承Hadoop提供的Reducer类，并重写reduce 方法，在reduce方法中编写自己的业务逻辑代码，处理完数据后也是通过context上下文对象将数据写出，这也就是最终的结果文件。

二、如何实现Hadoop中的序列化，以及Hadoop的序列化和Java的序列化有什么区别？

​ 首先序列化是把内存中的Java对象转化成二进制字节码，反序列化是将二进制字节码转化成Java对象，通常我们在对Java对象进行磁盘持久化写入或者将Java对象作为数据进行网络传输的时候需要进行序列化，相反如果要将J数据从磁盘读出并转化成Java对象需要进行反序列化。实现Hadoop中的序列化需要让JavaBean对象实现Writable接口，并重写write() 方法和readFields()方法，其中write()方法是序列化方法，readFields()方法是反序列化方法。

​ Hadoop序列化和Java序列化的区别在于，Java序列化更重量级，Java序列化的后的结果不仅仅生成二进制字节码文件，同时还会针对当前Java对象生成对应的检验信息以及集成体系结构，这样的话 无形中我们需要维护更多的数据，但是Hadoop序列化不会产生除了Java对象内部属性外的任何信息，整体内容更加简洁紧凑，读写速度相应也会提升很多，这也符合了大数据的处理背景。

三、概述一下MR程序的执行流程

​ 简单的描述，MR程序执行先从InputFormat类说起，由InputFormat负责数据读入，并在内部实现切片，每一个切片的数据对应生成一个MapTask任务，MapTask中按照文件的行逐行数据进行处理，每一行数据会调用一次我们自定义的Mapper类的map方法，map方法内部实现具体的业务逻辑，处理完数据会通过context对象将数据写出到磁盘（此处会经历Shuffle过程，详情请参考下面第七问！！！），接下来ReduceTask会开始执行，首先ReduceTask会将MapTask处理完的数据结果拷贝过来，每一组相同key的values会会调用一次我们自定的Reducer类的reduce方法，当数据处理完成后，会通过context对象将数据结果写出到磁盘上。

四、InputFormat负责数据写的时候要进行切片，为什么切片大小默认是128M

​ 首先切片大小是可以通过修改配置参数来改变的，但是默认情况下是和切块blocksize大小一致，这样做的目的就是为了在读取数据的时候正好能一次性读取一个块的数据，避免了在集群环境下发生跨机器读取的情况，如果跨机器读取会造成二外的网络IO，不利于MR程序执行效率的提升。

​

五、描述一下切片的逻辑（从源码角度描述）

​ MR中的切片是发生在数据读入的阶段中，所以我们要关注InputFormat的实现，通过追溯源码，在InputFormat这个抽象类中有一个getSplits(),这个方法就是我们实现切片的具体逻辑。首先我们先关注两个变量，分别是 minSize 和 maxSize，通过对源码的跟踪默认情况 minSize = 1，maxSize = Long.MAX_VALUE，源码中声明了一个集合List splits = new ArrayList();，用于装载将来的切片对象并返回。接下来我们根据提交的job信息获取到当前要进行切片的文件详情，首先判断点前文件是否可以进行切分，这一步主要考虑到一些不支持切分的压缩文件时不能进行切片操作，否则就破坏了数据的完整性，如果当前文件可以切片的话，那么接下来就要计算切片的大小，计算切片大小一共需要三个因子，分别是minSize 、maxSize 、blocksize ，最后通过Math.max(minSize, Math.min(maxSize, blockSize)); 计算逻辑获取到切片大小，默认情况切片大小和数据库块大小一致，如果我们想改变切片大小可以通过修改一下两个配置参数实现 mapreduce.input.fileinputformat.split.minsize mapreduce.input.fileinputformat.split.maxsize，

如果把切片大小调大改mapreduce.input.fileinputformat.split.minsize
如果把切片大小调小改mapreduce.input.fileinputformat.split.maxsize。

当我们可以获取到切片大小后就可以继续往下执行，在最终完成切片之前还有一个关键判断，就是判断剩余文件是否要继续进行切片，如果剩余文件/切片大小>1.1 那就继续切片，否则就不会再进行切片，这个规则考虑的情况就就是让将来的切片尽可能资源使用均衡，不至于很小的文件内容也开启一个MapTask。到此整个切片规则就表述完毕了！

六、CombineTextInputFormat机制是怎么实现的

​ CombineTextInputFormat也是InputFormat的一个实现类，主要用于解决小文件场景的。如果我们在处理大量小文件的时候由于默认的切片规则是针对文件进行切片，所以就导致大量MapTask的产生 但是每一个MapTask处理的文件又很小 这样就违背了MapReduce的设计初衷。如果遇到以上场景我们就不能使用默认的切片规则了，而是使用CombineTextInputFormat中的切片规则。CombineTextInputFormat中的切片规则大概思路是 先在Job提交的时候设定一个参数为切片最大的值，当这个值设置好以后并且在Job提交中指定使用InputFormat的实现类为CombineTextInputFormat，那么接下来在切片的过程中首先会把当前文件的大小和设置的切片的最大值进行比较，如果小于切片的最大值那就单独划分为一块，如果大于切片的最大值并且小于两倍的切片的最大值那就把当前文件一分为二划分成两个块，以此类推逐个对文件进行处理，这个过程称之为虚拟过程。最后生成真正的切片的时候 根据虚拟好的文件进行合并，只要合并后文件大小不超过最开始设置好的切片的最大值那就继续追加合并文件直到达到设置好的切片的最大值，此时就会产生一个切片对应生成一个MapTask。

七、阐述一下 Shuffle机制流程

​ shuffle是MR执行过程中很重要，也是必不可少的一个过程。当MapTask执行完map() 方法后通过context对象写数据的时候开始执行shuffle过程。首先数据先从Map端写入到环形缓冲区内，写出的数据会根据分区规则进入到指定的分区，并且同时在内存中进行区内排序。环形缓冲区默认大小为100M,当数据的写入的容量达到缓冲区大小的80%，数据开始向磁盘溢写，如果数据很多的情况下 可能发生N次溢写，这样在磁盘上就会产生多个溢写文件，并且保证每个溢写文件中区内数据是有序的，接下来在磁盘中会把多次溢写的文件归并到一起形成一个文件，这归并的过程中会根据相同的分区进行归并排序，保证归并完的文件区内是有序的，到此shuffle过程在Map端就完成了。 接着Map端输出的数据会作为Reduce端的输入数据再次进行汇总操作，此时ReduceTask任务会把每一个MapTask中计算完的相同的分区的数据拷贝到ReduceTask的内存中，如果内存放不下，开始写入磁盘，再接着就是对数据进行归并排序，排完序还要根据相同key进行分组，将来一组相同的key对应的values调用一次reduce方法。如果有多个分区就会产生多个ReduceTask来处理，处理的逻辑都一样。

八、在MR程序中由谁来决定分区的数量，哪个阶段环节会开始往分区中写数据？

​ MR程序中，从编码设置的角度分析，在Job提交的时候可以设置ReduceTask的数量，ReduceTask的数量就决定的分区的编号，默认有多少ReduceTask任务就会产生多少个分区，但是具体应该设置多少ReduceTask是由具体的业务决定。在Map阶段的map方法中通过context.write()往出写数据的时候其实就往指定的分区中写数据了。

九、阐述MR中实现分区的思路（从源码角度分析）

​ 分区是MR中一个重要的概念，通常情况下 分区是由具体业务逻辑决定的，默认情况下不指定分区数量就会有一个分区，如果要指定分区我们可以通过在Job提交的时候指定ReduceTask的数量来指定的分区的数量。我们从Map端处理完数据后，数据就会被溢写到指定的分区中，而决定一个kv数据究竟写到哪个分区中是由Hadoop提供的分区器对象控制的，这个对象叫做Partitioner。Partitioner对象默认的实现HashPartitioner类，通过追溯源码 当我们调用map方法往出写数据的时候，会调用到HashPartitioner，它的规则就是用当前写出数据的key 和在Job提交中设置的ReducesTask的数量做取余运算，得到的结果就是当前数据要写入的分区的分区编号。 除此之外，我们也可以自定义分区器对象，需要继承Hadoop提供的Partitioner对象，然后重写getPartition() 方法，在该方法中根据自己的业务实现分区编号的返回。最后再将我们自定义的分区器对象设置到Job提交的代码中覆盖默认的分区规则。

十、Hadoop中实现排序的两种方案分别是什么

​ 第一种实现方式是 直接让参与比较的对象实现 WritableComparable 接口，并指定泛型，接下来
实现compareTo() 方法 在该方法中实现比较规则即可。

​ 第二种实现方式是 自定义一个比较器对象，需要继承WritableComparator类 ，重写它的compare方法，注意在构造器中调用父类对当前的要参与比较的对象进行实例化。注意当前要参与比较的对象必须要实现WritableComparable 接口。最后在Job提交代码中将自定义的比较器对象设置到Job中就可以了。

十一、描述一下Hadoop中实现排序比较的规则（源码角度分析）

​ Hadoop中的排序比较，本质上就是给某个对象获取到一个比较器对象，至于比较逻辑就直接调用该比较器对象中的compareTo() 方法来实现即可！接下来我们主要聊聊如何给一个对象获取比较器对象。从源码角度分析，我们在Hadoop的MapTask类中的init方法中关注一行代码comparator=job.getOutputKeyComparator(); 此代码就是在获取比较器对象，跟到该方法中，首先源码中会先从Job配置中获取是否指定过自定义的比较器对象，如果获取到已经设置的比较器对象的Class文件，接下来就会利用反射将比较器对象创建出来，到此获取比较器对象的流程就结束了。

​ 如果我们没有在Job中设置自定义的比较器对象，那就没办法用反射的形式获取比较器对象，接下来Hadoop框架会帮助我们创建，具体思路如下：

1.在获取之前有个前提 判断当前job的MapOutputKeyClass 是否实现了WritableComparable接口，因为我们 实现比较的时候是根据key进行比较的，所以重点关注MapOutputKeyClass。

2.如果上面一步正常，MapOutputKeyClass实现了WritableComparable接口，接下来考虑到参与比较的对象是Hadoop自身的数据类型，例如Text 、LongWritable 等，这些数据类型在类加载的时候就已将获取到了比较器对象，并且以当前对象的class为key,一当前对象的比较器对象为value维护在内存中的一个叫做comparators的HashMap中。所以从源码上看会执行

WritableComparator comparator = comparators.get(c); 这样以及代码，意思就是从comparators这个HashMap中根据当前对象的class文件获取它的比较器对象。如果不出意外到这就可以获取Hadoop自身数据类型对象的比较对象了。但是考虑的一些异常情况，比如内存溢出导致GC垃圾回收这时候可能会获取不到比较器对象，那么接下来会执行forceInit(c);这样一个方法，让类重新再加载一遍，确保万无一失。如果到此还是获取不到比较器对象，那么只有一种情况了，那就是当前的参与比较的对象不是Hadoop自身的数据类型，而是我们自定义的对象，在源码中也已看到 最后执行了一句

comparator = new WritableComparator(c, conf, true); 代码，意思是Hadoop会给我创建一个比较器对象。以上就是Hadoop中获取比较器对象的全过程了！

十二、编写MR的时候什么情况下使用Combiner ,具体实现流程是什么？

​ Combiner流程再MR中是一个可选流程，通常也是一种优化手段，当我们执行完Map阶段的计算后数据量比较大，kv组合过多。这样在Reduce阶段执行的时候会造成拷贝大量的数据以及汇总更多的数据。为了减轻Reduce的压力，此时可以选择在Map阶段进行Combiner操作，将一些汇总工作提前进行，这样会减少kv的数量从而在数据传输的过程中对IO的消耗也大大降低。

​ 实现Combiner的大概流程为：首先需要自定义一个Combiner类，接着继承Hadoop的Reducer类，重写reduce()方法，在该方法中进行合并汇总。最后把自定义的Combiner类设置到Job中即可！

十三、OutputFormat自定义实现流程描述一下

​ OutputFormat类是MR中最后一个流程了，它主要负责数据最终结果的写出，一般我们不需要自定义，默认即可。但是如果我们对象最终输出结果文件的名称或者输出路径有个性化的要求，就可以通过自定OutputFormat来实现。实现流程大概如下：

​ 首先自定义个OutputFormat类，然后继承OutputFormat，重写OutputFormat的getRecordWriter()方法，在该方法中返回 RecordWriter 对象。由于RecordWriter 也是Hadoop内部对象，如果我们想实现自己的逻辑，我们还得自定义个RecordWriter类，然后继承RecordWriter类，重写该类中的write() 方法和close()方法,在write() 方法中实现数据写出的逻辑，在close()方法对资源进行关闭。

十四、MR实现 ReduceJoin 的思路，以及ReduceJoin方案有哪些不足？

​ 说到MR中的ReduceJoin，首先在Map阶段我们对需要jion的两个文件的数据进行统一搜集，用一个对象去管理，另外还要再改对象中新增一个属性用于记录每一条数据的来源情况。当在Map阶段将数据搜集好后，直接写出，写出的时候一定要注意输出的key的选择，这个key一定两个文件的关联字段。接下来Reduce阶段开始执行，先把Map端处理完的数据拷贝过来，然后一组相同key的values就会进入到reduce() 方法,由于之前已经定义好key是两文件的关联数据，那本次进入reduce() 的做join就可以，第一步根据不同的数据来源将两文件的数据分别利用容器或者对象维护起来，然后遍历其中一个容器根据具体业务逻辑将想要关联的数据获取到即可 然后输出结果。

​ 以上就是ReduceJoin的大概思路，但是ReduceJoin相对来说比较耗费性能，而且如果出现数据倾斜场景，更不太好处理。

十五、MR实现 MapJoin 的思路，以及MapJoin的局限性是什么？

​ MapJoin顾名思义就是在Map端直接进行Join，不会走Reduce阶段，这样就很大程度的提升了MR的执行效率，同时也解决的数据倾斜给Reduce阶段带来的问题。接下来就聊聊MapJoin的核心思路，首先MapJoin的前提就是我们面对需要join的两个文件符合一个是大文件一个是小文件的条件。再次前提下，我们可以将小的文件提前缓存的内存中，然后让Map端直接处理大文件，每处理一行数据就根据当前的关联的字段到内存中获取想要的数据，然后将结果写出。

第三部分：YARN相关问题

一、MR中如何提交一个Job?

​ 执行一个MR的时候，首先从提交Job开始，通过跟踪Job提交源码流程，发现底层是通过一个叫做submitJobInternal 的内部方法完成提交，提交的大概步骤是 先检测输入输出路径是否正确合法，如果路径有问题立刻抛出异常、如果路径检测没有问题 下一步会开始对文件计算切片信息，根据切片信息确定最终要开启的MapTask的数量，再接下来会检测是否有分布式缓存文件，如果有的话会设置到 Job 中。以上都完事后，程序会自动将当前运行的jar包、切片信息、以及配置信息都放入到临时产生保存job信息的文件夹下，一切准备就绪后，提交Job就完成了，接下进入Job的执行流程。

​

二、描述一下YARN的工作机制

​ 客户端向Yarn中提交一个MR(MapReduce)任务,由ResourceManager接收处理，ResourceManager负责资源的分配, 根据任务计算出所需要的资源,如cpu资源和内存资源,将这些资源封装成Container容器对象。ResourceManager将任务和Container发送给Resource Scheduler(资源调度器)，Resource Scheduler将任务和Container分配给Application Master, 由Application Master进行二次划分, 将任务分解成MapTask和ReduceTask，Application Master将MapTask和ReduceTask分配给NodeManager,三个NodeManager随机接收到MapTask或者ReduceTask , 由NodeManager负责任务的执行。执行的过程中，Application Master会对NodeManager的任务完成情况进行监控, 而Applications Manager会对NodeManager的任务资源使用情况进行监控，如果NodeManager上的任务执行成功,会把成功信息发送给Application Master和ResourceManager, 然后ResourceManager会进行资源的回收。如果NodeManager上的任务执行失败,会把失败信息发送给Application Master和ResourceManager, 然后ResourceManager仍然会进行资源的回收. 此时Application Master会向ResourceManager再次申请资源, 重新将这个任务分配给这个NodeManager , 循环往复, 直到任务执行成功（默认会最多反复执行4次）

三、YARN中有几种资源调度实现策略？分别阐述一下各自的特点

​ Yarn的资源调度策略一共有三种实现方式，首先是传统的单条FIFO队列，先进先出，当我们提交任务后 会把Job放入到FIFO队列中，然后按照谁先到谁先执行的原则运行，FIFO策略的特点就是需要排队，效率不高，而且遇到一些耗费时间长的Job会影响到后面Job的运行。其次就是容量调度策略，其实容量调度策略就是在单条FIFO的基础上，可以添加多个FIFO队列，这样就解决单条FIFO队列的不足，而且容量调度器还可以指定额定资源占比和最大资源占比以及还可以配置Job运行的优先级等，使用起来非常灵活。最后还有公平调度器策略，公平调度策略就是在容量调度策略的基础上进行了改造，主要特点就是队列中的所有Job在资源充足的情况下下可以并行运行。

四、简述一下如何配置一个自定的 队列？

​ 配置容量调度队列本质上就是修改Yarn的capacity-scheduler.xml配置文件，分别制定队列的名称，以及相关的属性配置，其中重点就是配置额定容量和最大容量，以及队里中运行的Job的优先级等...

第四部分：Hadoop优化和HA方面

一、Hadoop中的压缩作为一种常用的优化手段，经常被用在什么场景下？

​ 一般使用压缩，其实考虑的出发点无非就是想让文件变得小一点，在传输过程中对IO的消耗更小一些。我们在MR过程中大概率会在Mapper阶段结束的的时候对落盘的结果文件进行压缩，这样在Reduce阶段拷贝数据的时候就会轻松很多，从而提升了整个MR的执行效率。其次也可在MR开始执行的时候读取数据的时候直接读取压缩好的文件，这样效率也会有所提升，这在此环节要注意切片问题，如果选用的是一些不支持切片的压缩文件那就把文件大小控制在130M以内。最后再Reduce阶段结束的时候也可以进行压缩，在这压缩主要考虑将来对存储结果数据的时候对磁盘使用率的提升。

二、如果想要使用压缩，Hadoop如何对某一种压缩编码格式进行取舍？

​ 针对压缩编码格式的选择，大概从三个维度考虑即可，首先是压缩率，其次是压缩速度，再其次是压缩后是否支持切片，如果再考虑一种情况那就是这个压缩编码格式是否是Hadoop自带的。

三、你们公司常用的压缩方式有哪些？

​ 我们公司之前对于一些冷数据集采取过压缩，考虑到不经常使用，长久存储的情况，使用了Bzip2压缩方式。在MR中Mapper阶段落盘文件压缩的场景下我们使用的是Snappy,主要发挥它的压缩和解压缩速度。

四、从哪些方面定位MR执行的效率（如何分析MR执行慢的原因）

​ 决定MR的执行效率因素有很多，从大的层面划分的话，要从硬件和软件两方面。

硬件方面很好处理，其实说白了就是给服务器提升配置，花钱就能搞定。软件方面的话就比较复杂一些了，我们通常考虑小文件场景、数据倾斜问题、大量不可切片的超大压缩文件、环形缓冲区大小比例设置不合理，合并次数过多等等因素去分析。

五、如果想对MR程序进行优化，应该从哪些方面入手以及可能用到的优化手段？

​ 真正想要对MR进行优化，一定要非常了解MR执行的每一个环节和流程。首先在数据输入阶段，如果我们处理的是大量的小文件，可以考虑提前把它们合并一下，或者在输入文件的时候才用 CombineTextInputFormat实现，来解决小文件场景。其次在Mapper阶段运行的时候，尽量减少溢写的次数，可以通过修改环形缓冲区的大小和后反向的触发比例来实现，也可以通过参数配置减少合并的次数，或者在不影响业务的情况下使用Combiner流程，来减少IO操作。如果运行到Reduce阶段，我们可以合理利用Reduce的Buffer，尽量的减少落盘操作，能够让一部分数据直接进入reduce方法去执行。综上所述，其实MR执行过程中IO和内存是非常珍贵的资源，我们想要优化必然从这两方面入手。

六、在Hadoop针对小文件的处理方案有哪些？

​ 在整个Hadoop的解决方案中，HDFS作为存储系统很害怕小文件，它的解决方案就是将文件通过一个MR程序进行归档管理，从而实现把批量小文件归档成一个文件，等使用的时候可以在解档出来。MR中如果遇到小文件可以使用CombineTextInputFormat实现来进行切片逻辑。或者Hadoop中也提供了uber模式实现JVM重用，但是这个方案很有局限性，一般我们也不用。当然我们如果遇到大量的小文件也可以在存储之前采用一些手段就进行合并处理。

七、如何解决MR中Reduce的数据倾斜问题？

​ 数据倾斜一般都是由差异化较大的业务所导致的。如果做Join的场景下解决数据倾斜最直接的方式就是不使用Reduce Join。如果是其他一些场景我们可以进一步对倾斜的业务数据进行抽样和范围分区，以及自己根据业务进行更合理的自定义分区。

八、大概简述一下 Hadoop每一代版本的新特性？

​ Hadoop1.x到Hadoop2.x的变化中最突出的就是 有了YARN架构，其次还实现了集群间的数据拷贝、回收站功能以及小文件存档的解决方案。到Hadoop3.x后主要的新特性就是多NN的出现，进一步巩固了HA高可用的实现，还有就是纠删码解决副本策略带来磁盘存储使用率过高的弊端。

九、什么是Hadoop的HA?

​ 所谓的HA就是 让Hadoop集群任何情况下都能正常提供服务。早在Hadoop2.x的版本就开始实现HA，只不过在此版本中我们可以搭建两台NN，一台为Active状态对外提供服务，一台处于StandBy状态充当替补，当Active状态的机器出问题StandBy机器马上切换成Active状态。在Hadoop3.x版本提出了多NN的实现方案。原理都是一样的。

十、如何实现HA的集群搭建?

​ 搭建Hadoop实现，本质上还是通过修改配置文件来实现，具体步骤如下：

1. 规划HA架构中的服务部署（以三台机器为例）

   hadoop102	hadoop103	hadoop104
   Namenode	Namenode	Namenode
   Datanode	Datanode	Datanode
   JournalNode	JournalNode	JournalNode
   ZKFC	ZKFC	ZKFC
   ZK	ZK	ZK
   ResourceManager	ResourceManager	ResourceManager
   NodeManager	NodeManager	NodeManager

   2. 搭建HDFS的HA

 2.1 修改配置文件 core-site.xml

 ```xml
 <configuration>
 		<!-- 把多个NameNode的地址组装成一个集群mycluster -->
 		<property>
 		  <name>fs.defaultFS</name>
 		  <value>hdfs://mycluster</value>
 		</property>
 
 		<!-- 指定hadoop数据的存储目录 -->
 		<property>
 			<name>hadoop.tmp.dir</name>
 			<value>/opt/module/ha/hadoop-3.1.3/data</value>
 		</property>
 
 		<!-- 配置HDFS网页登录使用的静态用户为atguigu -->
 		<property>
 			<name>hadoop.http.staticuser.user</name>
 			<value>atguigu</value>
 		</property>
 
 		<!-- 配置该atguigu(superUser)允许通过代理访问的主机节点 -->
 		<property>
 			<name>hadoop.proxyuser.atguigu.hosts</name>
 			<value>*</value>
 		</property>
 		<!-- 配置该atguigu(superUser)允许通过代理用户所属组 -->
 		<property>
 			<name>hadoop.proxyuser.atguigu.groups</name>
 			<value>*</value>
 		</property>
 		<!-- 配置该atguigu(superUser)允许通过代理的用户-->
 		<property>
 			<name>hadoop.proxyuser.atguigu.groups</name>
 			<value>*</value>
 		</property>
     </configuration>
 ```

 2.2 修改配置文件 hdfs-site.xml

 ```xml
 <configuration>
 		<!-- NameNode数据存储目录 -->
 		<property>
 			<name>dfs.namenode.name.dir</name>
 			<value>file://${hadoop.tmp.dir}/name</value>
 		</property>
 		<!-- DataNode数据存储目录 -->
 		<property>
 			<name>dfs.datanode.data.dir</name>
 			<value>file://${hadoop.tmp.dir}/data</value>
 		</property>
 		<!-- JournalNode数据存储目录 -->
 		<property>
 			<name>dfs.journalnode.edits.dir</name>
 			<value>${hadoop.tmp.dir}/jn</value>
 		</property>
 		<!-- 完全分布式集群名称 -->
 		<property>
 			<name>dfs.nameservices</name>
 			<value>mycluster</value>
 		</property>
 		<!-- 集群中NameNode节点都有哪些 -->
 		<property>
 			<name>dfs.ha.namenodes.mycluster</name>
 			<value>nn1,nn2,nn3</value>
 		</property>
 		<!-- NameNode的RPC通信地址 -->
 		<property>
 			<name>dfs.namenode.rpc-address.mycluster.nn1</name>
 			<value>hadoop102:8020</value>
 		</property>
 		<property>
 			<name>dfs.namenode.rpc-address.mycluster.nn2</name>
 			<value>hadoop103:8020</value>
 		</property>
 		<property>
 			<name>dfs.namenode.rpc-address.mycluster.nn3</name>
 			<value>hadoop104:8020</value>
 		</property>
 		<!-- NameNode的http通信地址 -->
 		<property>
 			<name>dfs.namenode.http-address.mycluster.nn1</name>
 			<value>hadoop102:9870</value>
 		</property>
 		<property>
 			<name>dfs.namenode.http-address.mycluster.nn2</name>
 			<value>hadoop103:9870</value>
 		</property>
 		<property>
 			<name>dfs.namenode.http-address.mycluster.nn3</name>
 			<value>hadoop104:9870</value>
 		</property>
 		<!-- 指定NameNode元数据在JournalNode上的存放位置 -->
 		<property>
 			<name>dfs.namenode.shared.edits.dir</name>
 			<value>qjournal://hadoop102:8485;hadoop103:8485;hadoop104:8485/mycluster</value>
 		</property>
 		<!-- 访问代理类：client用于确定哪个NameNode为Active -->
 		<property>
 			<name>dfs.client.failover.proxy.provider.mycluster</name>
 			<value>org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider</value>
 		</property>
 		<!-- 配置隔离机制，即同一时刻只能有一台服务器对外响应 -->
 		<property>
 			<name>dfs.ha.fencing.methods</name>
 			<value>sshfence</value>
 		</property>
 		<!-- 使用隔离机制时需要ssh秘钥登录-->
 		<property>
 			<name>dfs.ha.fencing.ssh.private-key-files</name>
 			<value>/home/atguigu/.ssh/id_rsa</value>
 		</property>
 		  
 	  </configuration>
 ```

 2.3 启动HDFS的HA集群

 -  在102、103、104 各个JournalNode节点上，输入以下命令启动journalnode服务
        

    ```shell
      hdfs --daemon start journalnode
    ```

 -  在 hadoop102的 nn1 上，对其进行格式化，并启动

    ```shell
    hdfs namenode -format
    hdfs --daemon start namenode
    ```

 -  分别在 hadoop103的nn2 和 hadoop104的nn3上，同步nn1的元数据信息

    ```shell
    hdfs namenode -bootstrapStandby
    ```

 -  分别在 hadoop103上启动nn1 和 hadoop104上启动nn2

    ```shell
    hdfs --daemon start namenode
    ```

 -  通过web地址访问nn1 nn2 nn3 来测试集群是否正常使用

    http:hadoop102:9870 
    http:hadoop103:9870 
    http:hadoop104:9870

 -  在每台机器上启动DN

    ```shell
    hdfs --daemon start datanode
    ```

 -   将其中的一个nn切换成Active状态

    ```shell
    hdfs haadmin -getServiceState nn1
    ```

 2.4 实现自动故障转移

 -  在core-site.xml文件中增加

    ```xml
    <!-- 指定zkfc要连接的zkServer地址 -->
    	   <property>
    		 <name>ha.zookeeper.quorum</name>		 <value>hadoop102:2181,hadoop103:2181,hadoop104:2181</value>
    	   </property>
    ```

 -  在hdfs-site.xml中增加

    ```xml
    <!-- 启用nn故障自动转移 -->
    	   <property>
    		  <name>dfs.ha.automatic-failover.enabled</name>
    		  <value>true</value>
    	   </property>
    ```

 -  修改后分发配置文件

    ```shell
    xsync /opt/module/ha/hadoop-3.1.3/etc/hadoop 
    ```

 -  关闭HDFS集群

    ```shell
    stop-dfs.sh
    ```

 -  启动Zookeeper集群

    ```shell
    zk.sh start
    ```

 -  初始化HA在Zookeeper中状态

    ```shell
    hdfs zkfc -formatZK
    ```

 -  启动HDFS服务

    ```shell
    start-dfs.sh
    ```

	3. 搭建YARN的HA高可用集群

 3.1 修改yarn-site.xml

 ```xml
 <!-- 启用resourcemanager ha -->
 		<property>
 			<name>yarn.resourcemanager.ha.enabled</name>
 			<value>true</value>
 		</property>
 	 
 		<!-- 声明三台resourcemanager的地址 -->
 		<property>
 			<name>yarn.resourcemanager.cluster-id</name>
 			<value>cluster-yarn1</value>
 		</property>
 		<!--指定resourcemanager的逻辑列表-->
 		<property>
 			<name>yarn.resourcemanager.ha.rm-ids</name>
 			<value>rm1,rm2,rm3</value>
 		</property>
 		
 		<!-- ========== rm1的配置 ========== -->
 		<!-- 指定rm1的主机名 -->
 		<property>
 			<name>yarn.resourcemanager.hostname.rm1</name>
 			<value>hadoop102</value>
 		</property>
 		<!-- 指定rm1的web端地址 -->
 		<property>
 			<name>yarn.resourcemanager.webapp.address.rm1</name>
 			<value>hadoop102:8088</value>
 		</property>
 		<!-- 指定rm1的内部通信地址 -->
 		<property>
 			<name>yarn.resourcemanager.address.rm1</name>
 			<value>hadoop102:8032</value>
 		</property>
 		<!-- 指定AM向rm1申请资源的地址 -->
 		<property>
 			<name>yarn.resourcemanager.scheduler.address.rm1</name>  
 			<value>hadoop102:8030</value>
 		</property>
 		<!-- 指定供NM连接的地址 -->  
 		<property>
 			<name>yarn.resourcemanager.resource-tracker.address.rm1</name>
 			<value>hadoop102:8031</value>
 		</property>
 		
 		<!-- ========== rm2的配置 ========== -->
 		<!-- 指定rm2的主机名 -->
 		<property>
 			<name>yarn.resourcemanager.hostname.rm2</name>
 			<value>hadoop103</value>
 		</property>
 		<property>
 			<name>yarn.resourcemanager.webapp.address.rm2</name>
 			<value>hadoop103:8088</value>
 		</property>
 		<property>
 			<name>yarn.resourcemanager.address.rm2</name>
 			<value>hadoop103:8032</value>
 		</property>
 		<property>
 			<name>yarn.resourcemanager.scheduler.address.rm2</name>
 			<value>hadoop103:8030</value>
 		</property>
 		<property>
 			<name>yarn.resourcemanager.resource-tracker.address.rm2</name>
 			<value>hadoop103:8031</value>
 		</property>
 		
 		<!-- ========== rm3的配置 ========== -->
 		<!-- 指定rm3的主机名 -->
 		<property>
 			<name>yarn.resourcemanager.hostname.rm3</name>
 			<value>hadoop104</value>
 		</property>
 		<property>
 			<name>yarn.resourcemanager.webapp.address.rm3</name>
 			<value>hadoop104:8088</value>
 		</property>
 		<property>
 			<name>yarn.resourcemanager.address.rm3</name>
 			<value>hadoop104:8032</value>
 		</property>
 		<property>
 			<name>yarn.resourcemanager.scheduler.address.rm3</name>
 			<value>hadoop104:8030</value>
 		</property>
 		<property>
 			<name>yarn.resourcemanager.resource-tracker.address.rm3</name>
 			<value>hadoop104:8031</value>
 		</property>
 	 
 		<!-- 指定zookeeper集群的地址 --> 
 		<property>
 			<name>yarn.resourcemanager.zk-address</name>
 			<value>hadoop102:2181,hadoop103:2181,hadoop104:2181</value>
 		</property>
 
 		<!-- 启用自动恢复 --> 
 		<property>
 			<name>yarn.resourcemanager.recovery.enabled</name>
 			<value>true</value>
 		</property>
 	 
 		<!-- 指定resourcemanager的状态信息存储在zookeeper集群 --> 
 		<property>
 			<name>yarn.resourcemanager.store.class</name>     
 			<value>org.apache.hadoop.yarn.server.resourcemanager.recovery.ZKRMStateStore</value>
 		</property>
 ```

 3.2 将yarn-site.xml文件进行分发

 ```shell
 xsync /opt/module/ha/hadoop-3.1.3/etc/hadoop/yarn-site.xml
 ```

 3.3 在任意的机器上启动yarn 
 start-yarn.sh

 3.4 通过访问web地址验证

 3.5 测试Yarn故障自动转移

十一、HDFS实现自动故障转移原理？

​ 当HDFS的HA集群启动的时候，当前每一台NN都会绑定一个ZKFC服务，充当自己的经纪人的角色，由ZKFC负责访问ZK集群，在ZK集群中创建一个临时的数据节点，当前有哪一台NN的ZKFC先在ZK上创建了该临时节点就相当于当前NN就是对外提供服务的Active状态的服务了，其他自动成为StandBy状态。接下来当HDFS的HA集群中的Active状态的NN不可用时，此时当前服务的ZKFC线程就会检测NN的健康状态，发现它不可用时会马上和ZK断开连接，随之由它创建的临时节点也就消失了，接下来通过ZK的通知机制原理，另外两台StandBy机器马上知晓ZK的数据节点有变化，此时两台StandBy状态的机器会同时在ZK创建临时节点，谁先创建好那么它就是Active状态的机器了，但是为了防止脑裂问题，在该节点正式成为Active状态之前会远程登录到之前为Active状态的机器上执行kill命令，强行把之前Active状态的机器杀死。然后自己正常对外提供服务。

​

​

标签：总结,切片,文件,对象,Hadoop,学习,MR,数据
From： https://www.cnblogs.com/HYBG/p/17156779.html