cmcdh集群管理课程培训

1、大数据课程之MapReduce(2) 培训讲师：吴超 课程安排标准和自定义计数器*Combiner和Partitioner编程*自定义排序和分组编程*表连接*常见的调度器*MR优化方法*常见的MapReduce算法*-加深拓展-常见大数据处理方法*了解计数器hadoop计数器:可以让开发人员以全局的视角来审查程序的运行情况以及各项指标，及时做出错误诊断并进行相应处理。内置计数器（MapReduce相关、文件系统相关和作业调度相关） .也可以通过 查看Hello you, hello me的计数器信息 Counters: 19 File Output Format Counters Bytes

2、Written=19/reduce输出到hdfs的字节数 FileSystemCounters FILE_BYTES_READ=481 HDFS_BYTES_READ=38 FILE_BYTES_WRITTEN=81316 HDFS_BYTES_WRITTEN=19 File Input Format Counters Bytes Read=19/map从hdfs读取的字节数 Map-Reduce Framework Map output materialized bytes=49 Map input records=2 /map读入的记录行数 Reduce shuffle bytes=0 S

3、pilled Records=8 Map output bytes=35 Total committed heap usage (bytes)=266469376 SPLIT_RAW_BYTES=105 Combine input records=0 Reduce input records=4/reduce从map端接收的记录行数 Reduce input groups=3/reduce函数接收的key数量，即归并后的k2数量 Combine output records=0 Reduce output records=3/reduce输出的记录行数 Map output records=4

4、/map输出的记录行数自定义计数器与实现Context类调用方法getCounter()org.apache.hadoop.mapreduce.Counter计数器声明1.通过枚举声明context.getCounter(Enum enum) 2.动态声明context.getCounter(String groupName,String counterName) 计数器操作counter.setValue(long value);/设置初始值counter.increment(long incr);/增加计数Partitioner是partitioner的基类，如果需要定制partition

5、er也需要继承该类。2. HashPartitioner是mapreduce的默认partitioner。计算方法是which reducer=(key.hashCode() & Integer.MAX_VALUE) % numReduceTasks，得到当前的目的reducer。3. (例子以jar形式运行) Partitioner编程在map和reduce阶段进行排序时，比较的是k2。v2是不参与排序比较的。如果要想让v2也进行排序，需要把k2和v2组装成新的类，作为k2，才能参与比较。分组时也是按照k2进行比较的。排序和分组Combiners编程 每一个map可能会产生大量的输出，com

6、biner的作用就是在map端对输出先做一次合并，以减少传输到reducer的数据量。 combiner最基本是实现本地key的归并，combiner具有类似本地的reduce功能。 如果不用combiner，那么，所有的结果都是reduce完成，效率会相对低下。使用combiner，先完成的map会在本地聚合，提升速度。注意：Combiner的输出是Reducer的输入，Combiner绝不能改变最终的计算结果。所以从我的想法来看，Combiner只应该用于那种Reduce的输入key/value与输出key/value类型完全一致，且不影响最终结果的场景。比如累加，最大值等。Shuffle

7、每个map有一个环形内存缓冲区，用于存储任务的输出。默认大小100MB（io.sort.mb属性），一旦达到阀值0.8（io.sort.spill.percent）,一个后台线程把内容写到(spill)磁盘的指定目录（mapred.local.dir）下的新建的一个溢出写文件。写磁盘前，要partition,sort。如果有combiner，combine排序后数据。等最后记录写完，合并全部溢出写文件为一个分区且排序的文件。Reducer通过Http方式得到输出文件的分区。TaskTracker为分区文件运行Reduce任务。复制阶段把Map输出复制到Reducer的内存或磁盘。一个Map任务

8、完成，Reduce就开始复制输出。排序阶段合并map输出。然后走Reduce阶段。hadoop的压缩codecCodec为压缩，解压缩的算法实现。 在Hadoop中，codec由CompressionCode的实现来表示。下面是一些实现：压缩算法效率比较MapReduce的输出进行压缩输出的压缩属性：MapReduce中的join操作reduce side joinmap side joinreduce side joinreduce side join是一种最简单的join方式，其主要思想如下：在map阶段，map函数同时读取两个文件File1和File2，为了区分两种来源的key/valu

9、e数据对，对每条数据打一个标签（tag）,比如：tag=0表示来自文件File1，tag=2表示来自文件File2。即：map阶段的主要任务是对不同文件中的数据打标签。在reduce阶段，reduce函数获取key相同的来自File1和File2文件的value list， 然后对于同一个key，对File1和File2中的数据进行join（笛卡尔乘积）。即：reduce阶段进行实际的连接操作。map side join之所以存在reduce side join，是因为在map阶段不能获取所有需要的join字段，即：同一个key对应的字段可能位于不同map中。Reduce side join是

10、非常低效的，因为shuffle阶段要进行大量的数据传输。Map side join是针对以下场景进行的优化：两个待连接表中，有一个表非常大，而另一个表非常小，以至于小表可以直接存放到内存中。这样，我们可以将小表复制多份，让每个map task内存中存在一份（比如存放到hash table中），然后只扫描大表：对于大表中的每一条记录key/value，在hash table中查找是否有相同的key的记录，如果有，则连接后输出即可。为了支持文件的复制，Hadoop提供了一个类DistributedCache，使用该类的方法如下：（1）用户使用静态方法DistributedCache.addCach

11、eFile()指定要复制的文件，它的参数是文件的URI（如果是HDFS上的文件，可以这样：hdfs:/namenode:9000/home/XXX/file，其中9000是自己配置的NameNode端口号）。JobTracker在作业启动之前会获取这个URI列表，并将相应的文件拷贝到各个TaskTracker的本地磁盘上。（2）用户使用DistributedCache.getLocalCacheFiles()方法获取文件目录，并使用标准的文件读写API读取相应的文件。调度算法hadoop目前支持以下三种调度器：FifoScheduler：最简单的调度器，按照先进先出的方式处理应用。只有一个队列

12、可提交应用，所有用户提交到这个队列。没有应用优先级可以配置。CapacityScheduler：可以看作是FifoScheduler的多队列版本。每个队列可以限制资源使用量。但是，队列间的资源分配以使用量作排列依据，使得容量小的队列有竞争优势。集群整体吞吐较大。延迟调度机制使得应用可以放弃跨机器或者跨机架的调度机会，争取本地调度。 详情见官网 FairScheduler：多队列，多用户共享资源。特有的客户端创建队列的特性，使得权限控制不太完美。根据队列设定的最小共享量或者权重等参数，按比例共享资源。延迟调度机制跟CapacityScheduler的目的类似，但是实现方式稍有不同。资源抢占特性，

13、是指调度器能够依据公平资源共享算法，计算每个队列应得的资源，将超额资源的队列的部分容器释放掉的特性。 详情见官网 配置FairScheduler修改mapred-site.xml，然后重启集群更多配置见conf/fair-scheduler.xml配置CapacityScheduler修改mapred-site.xml，然后重启集群更多配置见conf/capacity-scheduler.xmlCapacityScheduler运行结果Hadoop StreamingHadoop Streaming是Hadoop提供的一个编程工具，它允许用户使用任何可执行文件或者脚本文件作为Mapper和Re

14、ducer，通过Hadoop Streaming编写的MapReduce应用程序中每个任务可以由不同的编程语言环境组成MapReduce的job如何调优如果存在大量的小数据文件，可以使用SequenceFile、自定义的CombineFileInputFormat推测执行在整个集群上关闭，特定需要的作业单独开启，一般可以省下约5%10%的集群资源 mapred.map.task.speculative.execution=true; mapred.reduce.task.speculative.execution=false; 开启jvm重用 mapred.job.reuse.jvm.num.tasks=-1 增加InputSplit大小 mapred.min.split.size=268435456增大map输出的缓存 io.sort.mb=300增加合并spill文件数量 io.sort.factor=50map端输出压缩，推荐LZO压缩算法 press.map.output=true; pression.lzo.LzoCodec; 增大shuffle复制线程数 mapred.reduce.parallel.