一. MapReduce编程模型还是用经典图片说明问题。
1 .首先，我们有输入，可以确认他的数据量很大
通过split后，他变成了几个分片，每个分片都处理成一个Map
3 .映射处理完成后，tasktracker会复制数据并对其排序，用输出的密钥和值进行分区，并将与分区相同的映射输出合并到同一reduce的输入中。 4. ruducer通过处理输出数据，每个相同的key必须用一个reduce完成处理， 让我们来看一下每个reduce对应于至少一个输出(可以通过扩展MultipleOutputFormat来获得多划分输出)5.一个例子。下面的图： (来自010-3010书中)几点输入到： ) 5.1中的德这里的例子是从日志文件中提取每年的每天气温。 最后计算的年最高气温5.3map的输出是每个key-value 5.4通过shuffle后的reduce输入。 迭代程序5.5reduce内置了对应于同一密钥的value，其任务是提取每年的最高气温。 然后，输出Mapper 1. Mapper可以选择性地继承名为MapreduceBase的基类。 他只是实现了一些方法，即使方法主体为空，也必须实现mapper接口(0.20之前的版本)。 这是通用接口，需要执行输入/输出的密钥值类型。这些类型通常是Wriable接口的实现类3。 实现map方法有四个参数。 前两个是输入的Key和value。 第三个参数是输出收集器，用于收集输出。 第四个是报告器，用于报告一些状态。 调试3.1 input的默认值为一行一行。每日记录放在value中，3.2output每次收集一条K-V记录，一个k可以支持多个value，reduce中有一个iterator 4.覆盖cocod Reducer 1. reduce还可以与外部资源交换数据，并继承名为MapreduceBase的基类。 其作用与映射器相同。 2.Reducer必须实现reducer接口。 此接口也是通用接口，它实现类似于mapper的3.Reduce方法。 这种方法也有四种：第一种是输入的密钥，第二种是输入的值的迭代器，可以遍历所有值。 这相当于一个清单。 outputCollector与map输出相同，每次收集的都是key-value的形式，report的作用与map相同.4.在新版本中hadoop已经将后面两个参数作为一个context 当然关于版本的接口也是兼容的。 0.19.x5。 覆盖configure方法，以覆盖与map中的相同的close方法。 可以进行reduce结束后的处理工作。 (clean up )Combiner 1.combiner的作用是：对map的输出进行一次计算，得到初步合并结果，减少reduce的计算压力。 2.combiner的创建方法与reduce相同。 他本来是reducer的实现类3。 当Reducer符合函数f(a，b )=f ) f(a )，f(a ) )时，combinner与Reduce相同。 例如，sum(a、b、c、d、e、f、g ) ) min.4 .创建正确的组合器将优化整个mapreduce程序的性能。 (特别是如果reduce是性能瓶颈。)。 5.combiner可以与reducer不同。 在Configuration 1.之后添加的属性的值将复盖先前定义的同名属性的值。 定义为final的属性(将finaltrue/final标记添加到属性定义中)。 不会被后面同名属性定义的值复盖。 3 .系统属性的优先级高于资源中定义的属性。 也就是说，System.setProperty )方法复盖资源文件中定义的属性的值。 4 .系统属性定义必须在资源文件中有适当的定义才能有效。 5 .优先级高于- d选项定义的属性资源文件中定义的属性。 6 .在设置6.runjobs1.inputsoutput1.1之前确定有无输入。 (如果不存在的话会发生错误，所以最好用程序来判断。 ) 1.2判断输出是否已经存在。 存在的话会出错。 ) 1.3养成良好的习惯。 )判断后执行。 2 )设置映射器、RR
inputformat & outputformat & types 3.1 input和output format都有两种, 一种是 textfile, 一种是sequencefile. 简单理解, textfile是文本组织的形式,sequence file是 二进制组织的形式. 3.2 Types的设置, 根据输入和输出的数据类型, 设置各种Writable接口的实现类的class对象. 4. 设置reduce count 4.1 reduce count可以为0, 着急的朋友的数据无需reduce的时候. 4.2 reduce数量最好稍微少于当前可用的slots的数量, 这样reduce就能在一波计算中算好. (一个slot可以理解为一个计算单元(资源).) 七. 其他的一些细节. 1. ChainMapper可以实现链式执行mapper 他本身就是一个Mapper的实现类. 提供了一个addMapper的方法. 2. ChainReducer 跟ChainMapper类似, 可以实现链式执行reducer, 他是Reducer的实现类. 3. 多个job先后运行, 可以通过先后执行 JobClient.runJob方法来实现先后顺序 4. 扩展MultipleOutputFormat接口, 可以实现一个reduce对应多份输出 (而且可以指定文件名哦) 5. Partitioner 接口用于将 Map的输出结果进行分区, 分区相同的key对应的数据会被同一个reducer处理 5.1 提供了一个接口方法: public int getPartition(K2 key, V2 value, int numReduceTasks) 5.2 可以自己定义, 根据key的某些特指来划分, 也可以根据value的某些特质来划分. 5.3 numReduceTasks就是设置的reduce的个数.一般返回的partition的值应该都小于这个值.(%) 6. reporter的作用 6.1 reporter.incrCounter(key, amount). 比如对数据计算是, 一些不合规范的脏数据, 我们可以通过counter来记录有多少 6.2 reporter.setStatus(status); 方法可以设置一条状态消息, 当我们发现job运行出现这条消息是, 说明出现了我们预期的(错误或者正确)的情况, 用于debug. 6.3 reporter.progress(), 像mapreduce框架报告当前运行进度. 这个progress可以起到心跳的作用. 一个task要是超过10分钟没有想mapreduce框架报告情况, 这个reduce会被kill掉. 着急的朋友的任务处理会比较旧是, 最好定时向mapreduce汇报你的状态. 7. 通过实现Wriable接口, 我们可以自定义key和value的类型, 使用起来就像pojo, 不需要每次都进行parse. 如果你的自定义类型是Key的类型, 则需要同时实现Comparable 接口, 用于排序. 比如MapWritable就是一个例子. 八. 实战.(简单篇)简单篇:
1. 需求: 统计某个站点每天的PV
2. 数据输入: 以天为分区存放着的日志数据, 一条日志代表一个PV
3. 数据输出: 日期 PV
4. Mapper编写
主要的工作很简单, split每一条日志, 取出日期, 并对该日期的PV搜集一条记录, 记录的value为ONE(1, 一条记录代表一个PV)
5. Reducer编写
reduce的任务是将每天(key相同的为同一天) 的日志进行汇总(sum), 最后以天为key输出汇总结果.
6. 设置环境, 指定job(Run)
6.1 设置输入路径.
6.2 设置输出路径
6.3 设置Mapper/Reducer 和 输入数据的数据格式和数据类型
6.4 执行命令:
hadoop jar site-pv-job.jar org.jiacheo.SitePVSumSampleJob
6.5 查看hadoop的web 工具, 显示当前job进度.
可以看出, 此次输入产生了14292个map,和29个reduce. reduce数这么少是因为我的reduce的slots数只有30, 所以设置为29, 以防一个挂了, 还能在一波reduce中算好.
6.6 计算结果.
上面部分是hadoop cli客户端显示的进度, 中间是web工具显示的输入输出的一些数据的统计.可以看出, 此次输入数据总共有1.6TB大小, 设计的总记录数为69.6亿. 也就是这份数据记录了该站点的69.6亿的PV. 左下角可以看出, 执行时间比较长, 用了18分钟+46秒.这里慢的原因不在于reduce, 而是我的map的slots太少, 只有300个, 总共一万多个map, 那要分好几百波才能算完map, 所以瓶颈在map这里.右下角是统计的结果数据, 可以看出, 该站点的整体的PV是呈现上升趋势的.
至此, 一个简单的map/reduce程序就写好并运行了.
下面介绍复杂一点的实践. 当然, 还是等有时间再来介绍吧. 碎觉先.