离线计算系统-(MAPREDUCE详解)v
目录课程大纲（MAPREDUCE详解）31. MAPREDUCE入门41.1 为什么要MAPREDUCE41.2 MAPREDUCE程序运行演示41.3 MAPREDUCE 示例编写及编程规范41.3.1 编程规范41.3.2 wordcount示例编写51.4 MAPREDUCE程序运行模式及debug方法71.4.1 本地运行模式71.4.2 集群运行模式72. Mapreduce程序的核心运行机制82.1 概述82.2 mr程序运行流程82.2.1 流程示意图82.2.2 流程解析82.3 Maptask实例数的决定机制102.3.1 maptask数量的决定机制10切片机制：102.4 ReduceTask实例数的决定113. MAPREDUCE中的Combiner124. MAPREDUCE中的序列化124.1 概述124.2 Jdk序列化和MR序列化之间的比较124.3 自定义对象实现MR中的序列化接口135. Mapreduce中的排序初步165.1 需求：165.2 分析165.3 实现166. Mapreduce中的分区Partitioner206.1 需求：206.2 分析206.3 实现207. mapreduce的shuffle机制227.1 概述：227.2 主要流程：227.3 详细流程227.4 详细流程示意图238. mapreduce数据压缩248.1 概述248.2 MR支持的压缩编码248.3 Reducer输出压缩248.4 Mapper输出压缩258.5 压缩文件的读取259. MapReduce与YARN279.1 YARN概述279.2 YARN的重要概念279.3 Yarn中运行运算程序的示例2710. MapReduce编程案例2810.1 reduce端join算法实现2810.2 map端join算法实现2910.3 web日志预处理32附：Mapreduce参数优化3611.1 资源相关参数3611.2 容错相关参数3711.3 本地运行mapreduce 作业3711.4 效率和稳定性相关参数37课程大纲（MAPREDUCE详解）MapReduce快速入门如何理解map、reduce计算模型Mapreudce程序运行演示Mapreduce编程规范及示例编写Mapreduce程序运行模式及debug方法MapReduce高级特性Mapreduce程序的核心机制MapReduce的序列化框架MapReduce的排序实现MapReduce的分区机制及自定义Mapreduce的数据压缩Mapreduce与yarn的结合Mapreduce编程案例Mapreduce 参数优化目标：掌握mapreduce分布式运算框架的编程思想掌握mapreduce常用算法的编程套路掌握mapreduce分布式运算框架的运行机制，具备一定自定义开发的能力1. MAPREDUCE原理篇（1）Mapreduce是一个分布式运算程序的编程框架，是用户开发“基于hadoop的数据分析应用”的核心框架；Mapreduce核心功能是将用户编写的业务逻辑代码和自带默认组件整合成一个完整的分布式运算程序，并发运行在一个hadoop集群上；1.1 为什么要MAPREDUCE（1）海量数据在单机上处理因为硬件资源限制，无法胜任（2）而一旦将单机版程序扩展到集群来分布式运行，将极大增加程序的复杂度和开发难度（3）引入mapreduce框架后，开发人员可以将绝大部分工作集中在业务逻辑的开发上，而将分布式计算中的复杂性交由框架来处理设想一个海量数据场景下的wordcount需求：单机版：内存受限，磁盘受限，运算能力受限分布式：1、 文件分布式存储（HDFS）2、 运算逻辑需要至少分成2个阶段（一个阶段独立并发，一个阶段汇聚）3、 运算程序如何分发4、 程序如何分配运算任务（切片）5、 两阶段的程序如何启动？如何协调？6、 整个程序运行过程中的监控？容错？重试？可见在程序由单机版扩成分布式时，会引入大量的复杂工作。为了提高开发效率，可以将分布式程序中的公共功能封装成框架，让开发人员可以将精力集中于业务逻辑。而mapreduce就是这样一个分布式程序的通用框架，其应对以上问题的整体结构如下：1、 MRAppMaster(mapreduce application master)2、 MapTask3、 ReduceTask1.2 MAPREDUCE框架结构及核心运行机制1.2.1 结构一个完整的mapreduce程序在分布式运行时有三类实例进程：1、MRAppMaster：负责整个程序的过程调度及状态协调2、mapTask：负责map阶段的整个数据处理流程3、ReduceTask：负责reduce阶段的整个数据处理流程1.2.2 MR程序运行流程1.2.2.1 流程示意图1.2.2.2 流程解析1、 一个mr程序启动的时候，最先启动的是MRAppMaster，MRAppMaster启动后根据本次job的描述信息，计算出需要的maptask实例数量，然后向集群申请机器启动相应数量的maptask进程2、 maptask进程启动之后，根据给定的数据切片范围进行数据处理，主体流程为：a) 利用客户指定的inputformat来获取RecordReader读取数据，形成输入KV对b) 将输入KV对传递给客户定义的map()方法，做逻辑运算，并将map()方法输出的KV对收集到缓存c) 将缓存中的KV对按照K分区排序后不断溢写到磁盘文件3、 MRAppMaster监控到所有maptask进程任务完成之后，会根据客户指定的参数启动相应数量的reducetask进程，并告知reducetask进程要处理的数据范围（数据分区）4、 Reducetask进程启动之后，根据MRAppMaster告知的待处理数据所在位置，从若干台maptask运行所在机器上获取到若干个maptask输出结果文件，并在本地进行重新归并排序，然后按照相同key的KV为一个组，调用客户定义的reduce()方法进行逻辑运算，并收集运算输出的结果KV，然后调用客户指定的outputformat将结果数据输出到外部存储1.3 MapTask并行度决定机制maptask的并行度决定map阶段的任务处理并发度，进而影响到整个job的处理速度那么，mapTask并行实例是否越多越好呢？其并行度又是如何决定呢？1.3.1 mapTask并行度的决定机制一个job的map阶段并行度由客户端在提交job时决定而客户端对map阶段并行度的规划的基本逻辑为：将待处理数据执行逻辑切片（即按照一个特定切片大小，将待处理数据划分成逻辑上的多个split），然后每一个split分配一个mapTask并行实例处理这段逻辑及形成的切片规划描述文件，由FileInputFormat实现类的getSplits()方法完成，其过程如下图：1.3.2 FileInputFormat切片机制1、切片定义在InputFormat类中的getSplit()方法2、FileInputFormat中默认的切片机制：a) 简单地按照文件的内容长度进行切片b) 切片大小，默认等于block大小c) 切片时不考虑数据集整体，而是逐个针对每一个文件单独切片比如待处理数据有两个文件：file1.txt 320Mfile2.txt 10M经过FileInputFormat的切片机制运算后，形成的切片信息如下： file1.txt.split1- 0128file1.txt.split2- 128256file1.txt.split3- 256320file2.txt.split1- 010M3、FileInputFormat中切片的大小的参数配置通过分析源码，在FileInputFormat中，计算切片大小的逻辑：Math.max(minSize, Math.min(maxSize, blockSize); 切片主要由这几个值来运算决定minsize：默认值：1 配置参数： maxsize：默认值：Long.MAXValue 配置参数：blocksize因此，默认情况下，切片大小=blocksizemaxsize（切片最大值）：参数如果调得比blocksize小，则会让切片变小，而且就等于配置的这个参数的值minsize （切片最小值）：参数调的比blockSize大，则可以让切片变得比blocksize还大但是，不论怎么调参数，都不能让多个小文件“划入”一个split选择并发数的影响因素：1、 运算节点的硬件配置2、 运算任务的类型：CPU密集型还是IO密集型3、 运算任务的数据量1.4 map并行度的经验之谈如果硬件配置为2*12core + 64G，恰当的map并行度是大约每个节点20-100个map，最好每个map的执行时间至少一分钟。l 如果job的每个map或者 reduce task的运行时间都只有30-40秒钟，那么就减少该job的map或者reduce数，每一个task(map|reduce)的setup和加入到调度器中进行调度，这个中间的过程可能都要花费几秒钟，所以如果每个task都非常快就跑完了，就会在task的开始和结束的时候浪费太多的时间。配置task的JVM重用JVM重用技术不是指同一Job的两个或两个以上的task可以同时运行于同一JVM上，而是排队按顺序执行。可以改善该问题：（mapred.job.reuse.jvm.num.tasks，默认是1，表示一个JVM上最多可以顺序执行的task数目（属于同一个Job）是1。也就是说一个task启一个JVM）l 如果input的文件非常的大，比如1TB，可以考虑将hdfs上的每个block size设大，比如设成256MB或者512MB1.5 ReduceTask并行度的决定reducetask的并行度同样影响整个job的执行并发度和执行效率，但与maptask的并发数由切片数决定不同，Reducetask数量的决定是可以直接手动设置：/默认值是1，手动设置为4job.setNumReduceTasks(4);如果数据分布不均匀，就有可能在reduce阶段产生数据倾斜注意： reducetask数量并不是任意设置，还要考虑业务逻辑需求，有些情况下，需要计算全局汇总结果，就只能有1个reducetask尽量不要运行太多的reduce task。对大多数job来说，最好rduce的个数最多和集群中的reduce持平，或者比集群的 reduce slots小。这个对于小集群而言，尤其重要。1.6 MAPREDUCE程序运行演示Hadoop的发布包中内置了一个hadoop-mapredu
