1、认识MapReduce编程模型主要内容lMapReduce编程模型简介lWordCount编程实例lHadoop MapReduce架构lMapReduce实战开发MapReduce编程模型简介l MapReduce是一种可用于数据处理的编程模型。该模型比较简单，但用于编写有用的程序并不简单。Hadoop可以运行由各种语言编写的MapReduce程序。例如：Java、Ruby、Python和C+语言等。最重要的是，MapReduce程序本质上是并行运行的，因此可以将大规模的数据分析任务交给任何一个拥有足够多机器的运行商。MapReduce的优势在于处理大规模数据集。MapReduce编程模型简
2、介l 1、从MapReduce自身的命名特点可以看出，MapReduce由两个阶段组成：Map和Reduce。用户只需map()和reduce()两个函数，即可完成简单的分布式程序设计。l 2、map()函数以key/value对作为输入，产生另外一系列key/value对作为中间输出写入本地磁盘。MapReduce框架会自动将这些中间数据按照key值进行聚合，且key值相同的数据被统一交给reduce()函数处理。l 3、reduce()函数以key及对应的value列表作为输入，经合并key相同的value值后，产生另外一系列key/value对作为最终输出写入HDFS。MapReduce
3、编程模型简介lMapReduce设计目的：易于编程 良好的扩展性 高容错性WordCount编程实例l Mapper类：public class WordMapper extends Mapper public static final IntWritable val=new IntWritable(1);public static final Text word=new Text();public void map(Object key,Text value,Context context)throws InterruptedException,IOException String line
4、=value.toString();String arr=line.split(t);for(String wd:arr)word.set(wd);context.write(word,val);WordCount编程实例l Reducer类 public class WordReducer extends Reducer public IntWritable val=new IntWritable();public void reduce(Text key,Iterable values,Context context)throws InterruptedException,IOExcept
5、ion int sum=0;for(IntWritable value:values)sum+=value.get();val.set(sum);context.write(key,val);WordCount编程实例lmain类：public class WordCount public static void main(String args)throws IOException,ClassNotFoundException,InterruptedException String intput=null;String output=null;if(null!=args&args.lengt
6、h=2)intput=args0;output=args1;Job job=new Job(new Configuration(),word count);/创建一个job /以jar包的形式运行job.setJarByClass(WordCount.class);/设置Mapper类和Reducer类job.setMapperClass(Mapper.class);job.setReducerClass(Reducer.class);WordCount编程实例 /设置输出的key/value的输出数据类型 job.setOutputKeyClass(Text.class);job.setOu
7、tputValueClass(IntWritable.class);/设置输入输出的格式FileInputFormat.addInputPath(job,new Path(intput);FileOutputFormat.setOutputPath(job,new Path(output);System.exit(job.waitForCompletion(true)?0:1);else System.err.println(wordcount );运行结果WordCount编程实例l 用户编写完MapReduce程序后，按照一定的规则指定程序的输入和输出目录，并提交到Hadoop集群中，作业
8、在Hadoop中的执行过程如图所示。Hadoop将输入数据切分成若干个输入分片(input split)，并将每个split交给一个Map Task处理；Map Task不断的从对应的split中解析出一个个key/value，并调用map()函数处理，处理完之后根据Reduce Task个数将结果分成若干个分区(partition)写到本地磁盘；同时，每个Reduce Task从每个Map Task上读取属于自己的那个partition，然后基于排序的方法将key相同的数据聚集在一起，调用reduce()函数处理，并将结果输出到文件中。WordCount编程实例l 流程图如下：Hadoop
9、MapReduce架构Hadoop MapReduce架构l1）Client用户编写的MapReduce程序通过Client提交到JobTracker端；同时，用户可通过Client提供的一些接口查看作业的运行状态。在Hadoop内部用“作业”（Job）表示MapReduce程序。一个MapReduce程序可对应若干个作业，而每个作业会被分解成若干个Map/Reduce任务（Task）。l2）JobTrackerJobTracke负责资源监控和作业调度。JobTracker 监控所有TaskTracker 与job的健康状况，一旦发现失败，就将相应的任务转移到其他节点；同时，JobTracke
10、r 会跟踪任务的执行进度、资源使用量等信息，并将这些信息告诉任务调度器，而调度器会在资源出现空闲时，选择合适的任务使用这些资源。在Hadoop 中，任务调度器是一个可插拔的模块，用户可以根据自己的需要设计相应的调度器。Hadoop MapReduce架构l 3）TaskTracker TaskTracker 会周期性地通过Heartbeat 将本节点上资源的使用情况和任务的运行进度汇报给JobTracker，同时接收JobTracker 发送过来的命令并执行相应的操作（如启动新任务、杀死任务等）。TaskTracker 使用“slot”等量划分本节点上的资源量。“slot”代表计算资源（CPU
11、、内存等）。一个Task 获取到一个slot 后才有机会运行，而Hadoop 调度器的作用就是将各个TaskTracker 上的空闲slot 分配给Task 使用。slot 分为Map slot 和Reduce slot 两种，分别供MapTask 和Reduce Task 使用。TaskTracker 通过slot 数目（可配置参数）限定Task 的并发度。Hadoop MapReduce架构l 4）Task Task 分为Map Task 和Reduce Task 两种，均由TaskTracker 启动。HDFS 以固定大小的block 为基本单位存储数据，而对于MapReduce 而言，
12、其处理单位是split。split 是一个逻辑概念，它只包含一些元数据信息，比如数据起始位置、数据长度、数据所在节点等。它的划分方法完全由用户自己决定。但需要注意的是，split 的多少决定了Map Task 的数目，因为每个split 只会交给一个Map Task 处理。Hadoop MapReduce架构l Map Task 执行过程如下图 所示。由该图可知，Map Task 先将对应的split 迭代解析成一个个key/value 对，依次调用用户自定义的map()函数进行处理，最终将临时结果存放到本地磁盘上，其中临时数据被分成若干个partition，每个partition 将被一个R
13、educe Task 处理。Hadoop MapReduce架构l Reduce Task 执行过程下图所示。该过程分为三个阶段：从远程节点上读取MapTask 中间结果（称为“Shuffle 阶段”）；按照key 对key/value 对进行排序（称为“Sort 阶段”）；依次读取，调用用户自定义的reduce()函数处理，并将最终结果存到HDFS 上（称为“Reduce 阶段”）。MapReduce实战开发l 数据源 sogou500w数据或sogou4000w数据l 数据字段描述 Time：用户访问时间 Uid：用户的id Keyword：访问的关键字 Rank：点击排名 Order：页
14、数 Url：网址条件过滤l 统计出搜索过包含有“仙剑奇侠传”内容的UID及搜索关键字记录详细代码见XjUid.javal rank2的所有UID及数量代码见UidByRank.java搜索过仙剑奇侠传内容的UID、搜索记录static class UidMap extends Mapper Text uid=new Text();protected void map(LongWritable key,Text value,org.apache.hadoop.mapreduce.Mapper.Context context)throws java.io.IOException,Interrupt
15、edException String lines=value.toString().split(t);if(lines!=null&lines.length=6)String kw=lines 2;if(kw.indexOf(“仙剑奇侠传)=0)uid.set(lines1);context.write(uid,new Text(kw);搜索过仙剑奇侠传内容的UID、搜索记录public static void main(String args)throws IOException,ClassNotFoundException,InterruptedException if(args.leng
16、th!=2&args=null)System.err.println(Please Iput Right Path!);System.exit(0);Configuration configuration=new Configuration();Job job=new Job(configuration,BaiduUid.class.getSimpleName();job.setJarByClass(BaiduUid.class);job.setInputFormatClass(TextInputFormat.class);job.setOutputFormatClass(TextOutput
17、Format.class);FileInputFormat.setInputPaths(job,new Path(args0);FileOutputFormat.setOutputPath(job,new Path(args1);job.setMapperClass(UidMap.class);job.setNumReduceTasks(0);job.setOutputKeyClass(Text.class);job.setOutputValueClass(Text.class);job.waitForCompletion(true);条件查询l上午7-9点之间，搜索过“赶集网”的用户 详细代
18、码请见GjTest.java本章小结l MapReduce主要分为input、splitting、Mapping、Shuffling、Reducing、Final reduce这几个阶段。l Hadoop MapReduce处理的数据一般位于底层分布式文件系统中。该系统往往将用户的文件切分为若干个固定大小的block存储到不同的节点上。默认情况下，MapReduce的每个Task处理一个block。l MapReduce主要由四个组件构成，分别是Client，JobTracker，TaskTracker和Task，它们共同保障一个作业的成功运行。作业l用MapReduce实现WordCount谢谢
