Hadoop知识总结

一、Hadoop组成：

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

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

二、HDFS组成原理：

1、HDFS的架构：

我们可以看到在这个架构中有一个NameNode和几个DataNode还有一个SecondaryNameNode，下面我们分别来阐述一下它们各自的作用.

2、NameNode：

（1）、加载fsimag和edits文件到内存中。

ps：fsimag文件是HDFS元信息文件，edits文件是记录操作日志文件。fsimage文件和edits文件在一起，记录了数据块的位置和冗余信息。
（2）、客户端向NameNode发起增删改查的请求。

（3）、NameNode记录操作到内存中，更新滚动日志。

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

3、DataNode：

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

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

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

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

4、SecondaryNameNode：

（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。

5、NameNode、JobTracker、DataNode、TaskTracker

（1）、hadoop的集群是基于master/slave模式，namenode和jobtracker属于master，datanode和tasktracker属于slave，master只有一个，而slave有多个。
（2）、SecondaryNameNode内存需求和NameNode在一个数量级上，所以通常secondary NameNode（运行在单独的物理机器上）和 NameNode 运行在不同的机器上。
（3）、JobTracker对应于NameNode,TaskTracker对应于DataNode。
（4）、DataNode和NameNode是针对数据存放来而言的。JobTracker和TaskTracker是对于MapReduce执行而言的。

ps：mapreduce中几个主要概念，mapreduce 整体上可以分为这么几条执行线索：jobclient，JobTracker与TaskTracker。
1）JobClient会在用户端通过JobClient类将已经配置参数打包成jar文件的应用存储到hdfs，并把路径提交到Jobtracker,然后由JobTracker创建每一个Task（即 MapTask 和 ReduceTask）并将它们分发到各个TaskTracker服务中去执行。
2）JobTracker是一master服务，软件启动之后JobTracker接收Job，负责调度Job的每一个子任务。task运行于TaskTracker上，并监控它们，如果发现有失败的task就重新运行它。一般情况应该把JobTracker 部署在单独的机器上。
3）TaskTracker是运行在多个节点上的slaver服务。TaskTracker主动与JobTracker通信，接收作业，并负责直接执行每一个任务。TaskTracker 都需要运行在HDFS的DataNode上。

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

三、HDFS上传下载原理：

1、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循环上传数据。

2、HDFS文件的下载：

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

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

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

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

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

（6）、在fsimage中查找。

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

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

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

四、HDFS的功能：

1、回收站。

（1）、[-expunge] 清空回收站。

2、配额。

（1）、名称配额。

（2）、空间配额。

3、快照。

（1）、一种备份手段。

4、安全模式。

5、HDFS的联盟（Federation）和HA的区别：

（1）、联盟是有两个NameNode，并且都是active状态；HA也有两个，不过其中有一个是standby一个是active状态。

（2）、HA需要借助ZK来实现，而Federation不需要，Federation能让缓存更多的元信息。

（3）、HA的好处是：①负载均衡。②失败迁移。

6、命令：①操作命令：hdfs dfs -xxx ②管理命令：hdfs dfsadmin -xxx

7、HDFS的数据压缩方法：

（1）、压缩方法有bzip2，gzip，lzo，snappy，一般采用的是snappy，压缩的的位置在：map的输入，mapper的输出，压缩reducer的输出。

8、HDFS小文件优化方法：

（1）、Hadoop Archive：将小文件放入HDFS块中的文件归档工具，减少NameNode内存的使用。

（2）、采用SequenceFile格式。

（3）、CombineFileInputFormat：将多个文件合并成一个单独的split。

（4）、JVM重用：一个map运行一个JVM，一个map运行完毕，jvm会继续运行其他map。

五、MapReduce细节：

1、MR中有两个非常重要的东西：①MapTask ②ReduceTask（后面细讲）

2、FileInputFormat切片机制：

（1）、在我们提交任务的时候，会有一个FileInputFormat.setInputPath(job，args[0])，我好奇看到了源码，源码中它是继承了一个InputFormat，在InputFormat这个抽象类中，有两个抽象方法：

（2）、getSplits作用：1）、找到你数据得存储目录。2）、开始遍历处理目录下的每一个文件。3）、遍历个文件：①获取文件的大小。②计算切片的大小（ps：默认情况下切片的大小等于blocksize的大小，开始切，形成第1个切片：0:128M 第2个切片128:256M 第3个切片256M:300M（每次切片时，都要判断切完剩下的部分是否大于块的1.1倍，不大于1.1倍就划分一块切片），将切片信息写到一个切片规划文件中，整个切片的核心过程在getSplit()方法中完成。数据切片只是在逻辑上对输入数据进行分片，并不会再磁盘上将其切分成分片进行存储。InputSplit只记录了分片的元数据信息，比如起始位置、长度以及所在的节点列表等。注意：block是HDFS上物理上存储的存储的数据，切片是对数据逻辑上的划分。）4）提交切片规划文件到yarn上，yarn上的MrAppMaster就可以根据切片规划文件计算开启maptask个数。）。

3、自定义InputFormat：

（1）、自定义个类继承FileInputFormat

（2）、改写RecordReader，实现一次读取一个完整文件封装k，v。

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

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

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

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

（4）、切片时针对每一个文件单独切片，不考虑数据集整体。
5、ReduceTask个数由什么决定：

（1）、一个job的reduceTask数量是通过job.setNumReduceTasks(x)来设置的;x 为reduce的数量。不设置的话默认为 1。

6、MR中的优化：

1）、MR程序的效率瓶颈在于：①计算机的性能，比如CPU，内存，网络等。②I/O操作优化。比如：数据倾斜，map和reduce设置不合理，reduce等待时间过长，小文件多，大量不可分块的超大文件，spill次数过多，merge次数过多。

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

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

六、MapReduce中的MapTask：

1、MapTask的工作流程：

Read阶段-->Map阶段-->Collect阶段-->Spill溢写阶段-->Combiner阶段

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

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

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

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

七、MapReduce中的ReduceTask：

1、ReduceTask的工作机制：

Copy阶段-->Merge阶段-->Sort阶段-->Reduce阶段

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

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

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

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

八、MapReduce中的Shuffle（转载）：

1、Shuffle机制：MapReduce保证每个reducer的输入都是按键有序排列的，系统执行排序的过程（即将map输出作为输入传给reducer）称为shuffle。

2、如何优化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）、排序的分类：

1）、部分排序：MR根据输入记录的键对数据集排序。保证输出的每个文件内部排序。

2）、全排序：可以使用一个分区来产生一个全局排序的文件，但这种方法效率在处理大文件的时候效率很低。替换方案：首先创建一系列排序好的文件；其次，串联这些文件；后，生成一个全局排序的文件。

3）、辅助排序（GroupingComparator分组）：对reduce阶段的数据根据某一个或几个字段进行分组。

4）、二次排序：在自定义排序过程中，如果compareTo中的判断条件为两个即为二次排序。

（2）、自定义排序：

1）、bean实现了WritableComparable接口重写compareTo方法，就可以实现排序。

2）、规则：按照key2排序。如果是数字按照升序排列；如果是字符串，按照字典顺序。

3）、对象排序：实现WritableComparable接口。ps：该对象必须时key2，必须实现了序列化。

（3）、排序发生的阶段：

1）、一个是在mapper中，在spill后partition之前。

2）、一个是在reducer中，在copy后reduce之前。

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、资源相关参数：

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数目, 默认值: 1
6）、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分配的小内存

2、容错相关参数：

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.”。

十一、MapReduce扯谈：

1、MR中出现单点负载过大，怎么负载均衡：可以用partitioner。

2、MR中怎么实现TopN的求解：可以自定义groupingcomparator，对结果进行大值排序，然后再reduce输出时，控制只输出前n个数。就达到了topn输出的目的。

3、如何使用MR实现两个表的join？
1）reduce side join : 在map阶段，map函数同时读取两个文件File1和File2，为了区分两种来源的key/value数据对，对每条数据打一个标签（tag）,比如：tag=0 表示来自文件File1，tag=2 表示来自文件File2。
2）map side join : Map side join 是针对以下场景进行的优化：两个待连接表中，有一个表非常大，而另一个表非常小，以至于小表可以直接存放到内存中。这样，我们可以将小表复制多份，让每个map task 内存中存在一份（比如存放到hash table 中），然后只扫描大表：对于大表中的每一条记录key/value，在hash table 中查找是否有相同的key 的记录，如果有，则连接后输出即可。
4、有可能使Hadoop任务输出到多个目录么？如果可以，怎么实现？

1）自定义outputFormat，改写recordWriter，具体改写输出数据的方法write。

十二、yarn详解：

yarn的基本组成：resourceManager、nodeManger、applicationMaster、container

1、resourceManager：主要负责资源的调度和应用程序的管理。

①调度器：调度器是将系统中的资源分配给各个正在运行的应用程序。

②应用程序管理：负责管理所有applicationMaster。

2、nodeManager：定时告诉resourceManger，node节点的资源使用情况；任务的启动和终止。

3、applicationMaster：向resourceManger请求资源，监听任务的执行进度。

4、container：资源的抽象（cpu，内存的信息），当applicationMaster向recourceManger请求资源的时候，就是以container抽象资源的形式返回，限制资源的使用情况。

5、执行流程：

6、三种资源调度器：

（1）、FIFO Scheduler：这是一种先进先出的队列，进行资源分配的时候，给队列上中上的应用进行分配资源，待到头上的应用需求满足后再给下一个分配，以此类推。

（2）、Capacity Scheduler：Capacity 调度器允许多个组织共享整个集群，每个组织可以获得集群的一部分计算能力。

（3）、Fair Scheduler：Fair调度器的设计目标是为所有的应用分配公平的资源。

总结：FIFO Scheduler：按照顺序执行，不适合共享集群。Capacity Sheduler：专门的队列来运行小任务，为小任务专门预设了一定的资源，导致大任务的执行时间回落后使用FIFO调度器的时间。Fair Scheduler：公平调度，每个来的任务都会获得的相等的资源。

7、yarn的优点：

（1）、解决了运行的用户程序于yarn框架完全解耦。

（2）、yarn上面可以运行各种分布式框架。

（3）、MR一旦运行失败，ResourceManager负责重新启动，多重启次数可由用户设置，默认是两次。

（4）、Task周期性的向MRAppMaster汇报心跳，一旦Task挂掉，则MRAppMaster将为之重新申请资源，默认是四次。

8、MR中的推测执行算法及其原理：

（1）、作业完成时间是由慢的Task所决定的。

（2）、推测执行机制：如果发现拖后退的任务，启动一个备份任务，谁先完成则采用谁。

（3）、不能使用推测执行机制的情况：①任务之间存在严重的负载倾斜。②特殊任务：比如向数据库中写数据。

更多文章见：

让我遇见你。​www.shiruiblog.cn