1.Wordcount示例编写:
MapReduce采用”分而治之”的思想,把对大规模数据集的操作,分发给一个主节点管理下的各个分节点共同完成,然后通过整合各
个节点的中间结果,得到最终结果。简单地说,MapReduce就是”任务的分解与结果的汇总”
需求:在一堆给定的文本文件中统计输出每一个单词出现的总次数
1.1.定义一个Mapper类:
package cn.gigdata.hdfs.mr; import java.io.IOException; import org.apache.hadoop.io.IntWritable; import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Mapper; /** * 参数信息: * 1.KEYIN: 默认情况下,是mr框架所读到的一行文本的起始偏移量,Long,在hadoop中用自己的序列化接口,不直接用Long,用LongWritable * 2.VALUEIN:默认情况下,是mr框架所读到的一行文本的内容,String,同上,用String的序列化接口Text * 3.KEYOUT:是用户自定义逻辑处理完成之后输出数据中的key,在此处是单词,String,同上,用Text * 4.VALUEOUT:是用户自定义逻辑处理完成之后输出数据中的value,在此处是单词次数,Integer,同上,用IntWritable * Map程序只写一份,但是可以被很多Map Task调用 */ public class WordCountMapper extends Mapper<LongWritable, Text, Text, IntWritable>{ /** * map阶段的业务逻辑就写在自定义的map()方法中 * maptask会对每一行输入数据调用一次我们自定义的map()方法 */ @Override protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { //将maptask传给我们的文本内容先转换成String String line = value.toString(); //根据空格将这一行切分成单词 String[] words = line.split(" "); //将单词输出为<单词,1> for(String word:words){ //将单词作为key,将次数1作为value,以便于后续根据单词进行数据分发,以便于相同单词会到相同的reduce task context.write(new Text(word), new IntWritable(1)); } } }
1.2.定义一个Reducer类:
package cn.gigdata.hdfs.mr; import java.io.IOException; import org.apache.hadoop.io.IntWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Reducer; /** * KEYIN, VALUEIN 对应 mapper输出的KEYOUT,VALUEOUT类型对应 * KEYOUT, VALUEOUT 是自定义reduce逻辑处理结果的输出数据类型: KEYOUT是单词 VLAUEOUT是总次数 */ public class WordcountReducer extends Reducer<Text, IntWritable, Text, IntWritable>{ /** * <angelababy,1><angelababy,1><angelababy,1><angelababy,1> * <hello,1><hello,1><hello,1><hello,1><hello,1><hello,1> * <banana,1><banana,1><banana,1><banana,1><banana,1><banana,1> * 入参key,是一组相同单词kv对应的key(以组为单位调一次)reduce传key的时候传的是组中的第一个 */ //生命周期:框架每传递进来一个kv 组,reduce方法被调用一次 @Override protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException { int count=0; /*Iterator<IntWritable> iterator = values.iterator(); while(iterator.hasNext()){ count += iterator.next().get(); }*/ for(IntWritable value:values){ count += value.get(); } context.write(key, new IntWritable(count));//默认将结果写入(HDFS)文件 } }
1.3.定义一个主类,用来描述job并提交job:
package cn.gigdata.hdfs.mr; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.IntWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Job; import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; /** * 相当于一个yarn集群的客户端 * 需要在此封装我们的mr程序的相关运行参数,指定jar包 * 最后提交给yarn * @author * */ //定义一个主类,用来描述job并提交job public class WordcountDriver { //把业务逻辑相关的信息(哪个是mapper,哪个是reducer,要处理的数据在哪里,输出的结果放哪里……)描述成一个job对象 //把这个描述好的job提交给集群去运行 public static void main(String[] args) throws Exception { if (args == null || args.length == 0) { args = new String[2]; args[0] = "hdfs://shizhan2:9000/wordcount/input/wordcount.txt"; args[1] = "hdfs://shizhan2:9000/wordcount/output8"; } Configuration conf = new Configuration(); Job job = Job.getInstance(conf); //指定本程序的jar包所在的本地路径 job.setJarByClass(WordcountDriver.class); //指定本业务job要使用的mapper/Reducer业务类 job.setMapperClass(WordCountMapper.class); job.setReducerClass(WordcountReducer.class); //指定mapper输出数据的kv类型 job.setMapOutputKeyClass(Text.class); job.setMapOutputValueClass(IntWritable.class); //指定最终输出的数据的kv类型 job.setOutputKeyClass(Text.class); job.setOutputValueClass(IntWritable.class); //指定job的输入原始文件所在目录 FileInputFormat.setInputPaths(job, new Path(args[0])); //指定job的输出结果所在目录 FileOutputFormat.setOutputPath(job, new Path(args[1])); //将job中配置的相关参数,以及job所用的java类所在的jar包,提交给YARN集群中去运行 /*job.submit();*/ boolean res = job.waitForCompletion(true); System.exit(res?0:1); } }
1.4.MapReduce程序运行模式
1.4.1.本地运行模式
(1)mapreduce程序是被提交给LocalJobRunner在本地以单进程的形式运行
(2)而处理的数据及输出结果可以在本地文件系统,也可以在hdfs上
(3)怎样实现本地运行?写一个程序,不要带集群的配置文件(本质是你的mr程序的conf中是否有
mapreduce.framework.name=local以及yarn.resourcemanager.hostname参数)
(4)本地模式非常便于进行业务逻辑的debug,只要在eclipse中打断点即可
如果在windows下想运行本地模式来测试程序逻辑,需要在windows中配置环境变量:
%HADOOP_HOME% = d:/hadoop-2.6.1
%PATH% = %HADOOP_HOME%in
并且要将d:/hadoop-2.6.1的lib和bin目录替换成windows平台编译的版本
1.4.2.集群运行模式
(1)将mapreduce程序提交给yarn集群resourcemanager,分发到很多的节点上并发执行
(2)处理的数据和输出结果应该位于hdfs文件系统
(3)提交集群的实现步骤:
A、将程序打成JAR包,然后在集群的任意一个节点上用hadoop命令启动
hadoop jar workcount.jar cn.gigdata.hdfs.mr.WordcountDriver /wordcount/input1 /wordcount/outputWord
B、直接在linux的eclipse中运行main方法
(项目中要带参数:mapreduce.framework.name=yarn以及yarn的两个基本配置)
C、如果要在windows的eclipse中提交job给集群,则要修改YarnRunner类
1.5.MapReduce程序编程规范:
(1)用户编写的程序分成三个部分:Mapper,Reducer,Driver(提交运行mr程序的客户端)
(2)Mapper的输入数据是KV对的形式(KV的类型可自定义)
(3)Mapper的输出数据是KV对的形式(KV的类型可自定义)
(4)Mapper中的业务逻辑写在map()方法中
(5)map()方法(maptask进程)对每一个<K,V>调用一次
(6)Reducer的输入数据类型对应Mapper的输出数据类型,也是KV
(7)Reducer的业务逻辑写在reduce()方法中
(8)Reducetask进程对每一组相同k的<k,v>组调用一次reduce()方法
(9)用户自定义的Mapper和Reducer都要继承各自的父类
(10)整个程序需要一个Drvier来进行提交,提交的是一个描述了各种必要信息的job对象
1.6.MapReduce程序在集群中运行时的大体流程: