MapReduce编写实现wordcount词频统计

p>首先编写WordCountDriver：

创新互联公司服务项目包括鄂温克网站建设、鄂温克网站制作、鄂温克网页制作以及鄂温克网络营销策划等。多年来，我们专注于互联网行业，利用自身积累的技术优势、行业经验、深度合作伙伴关系等，向广大中小型企业、政府机构等提供互联网行业的解决方案，鄂温克网站推广取得了明显的社会效益与经济效益。目前，我们服务的客户以成都为中心已经辐射到鄂温克省份的部分城市，未来相信会继续扩大服务区域并继续获得客户的支持与信任！

package com.jym.hadoop.mr.demo;

import java.io.IOException;

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.CombineTextInputFormat;

import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;

import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

/**

* 这个程序相当于一个yarn集群的客户端，

* 需要在此封装我们的mr程序的相关运行参数，指定jar包,

* 最后提交给yarn

* */

public class WordcountDriver

{

public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException

{

Configuration conf=new Configuration();

/*其实如果在本地运行MR程序其实不用配置下面的代码程序，在MR默认下就是本地运行*/

/**下面这段代码配置的是在本地模式下运行MR程序*/

/**是否运行为本地模式，就是看这个参数值是否为local，默认就是local;*/

//conf.set("mapreduce.framework.name", "local"); //在本地运行MR程序

//本地模式运行MR程序时，输入输出的数据可以在本地，也可以在hdfs上

//到底在哪里，就看以下两行配置用哪一行了，默认是“file:///”

/**conf.set("fs.defaultFS", "hdfs://hadoop1:9000");*/ //使用的是HDFS系统

//conf.set("fs.defaultFS", "file:///"); //使用的是本地Windows磁盘

/**运行集群模式，就是把程序提交到yarn中去运行

* 要想运行为集群模式，以下3个参数要指定为集群上的值

* */

conf.set("mapreduce.framework.name", "yarn");

conf.set("yarn.resourcemanager.hostname", "hadoop1");

conf.set("fs.defaultFS", "hdfs://hadoop1:9000");

Job job = Job.getInstance(conf);

/**要想在Windows的Eclipse上运行程序，并提交到hadoop的YARN集群上需要指定jar包，如下：*/

/**job.setJar("c:/wc.jar");*/

//job.setJar("/home/hadoop/wc.jar"); //这种是将程序打包成jar包，放到指定的位置，缺乏灵活性，不建议使用;

//指定本程序的jar包所在的本地路径

job.setJarByClass(WordcountDriver.class);

//指定本业务job要使用的mapper/reducer业务类

job.setMapperClass(WordcountMapper.class);

job.setReducerClass(WordcountReducerr.class);

//指定mapper输出数据的kv类型;

job.setMapOutputKeyClass(Text.class);

job.setMapOutputValueClass(IntWritable.class);

//指定最终输出的数据的kv类型

job.setOutputKeyClass(Text.class);

job.setOutputValueClass(IntWritable.class);

//指定需要使用的combiner，以及用哪一个类作为combiner的逻辑

/*job.setCombinerClass(WordcountCombiner.class);*/

job.setCombinerClass(WordcountReducerr.class);

/**因为combiner的工作原理通reducecer的作用是一样的，所以直接反射调用reducerr类其实作用是一样的*/

/**此处为之后为测试添加的*/

//如果不设置InputFormat,它默认使用的是TextInputFormat.class

/**job.setInputFormatClass(CombineTextInputFormatInputFormatInputFormat.class);

CombineTextInputFormat.setMaxInputSplitSize(job, 4194304);

CombineTextInputFormat.setMinInputSplitSize(job, 2097152);

*/

//指定job的输入原始文件所在目录

//FileInputFormat.setInputPaths(job, new Path("/wordcount/input")); //此处添加的路径为HDFS文件系统的路径;

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);

}

}

其次编写WordCountMapper:

package com.jym.hadoop.mr.demo;

import java.io.IOException;

import org.apache.hadoop.io.IntWritable;

import org.apache.hadoop.io.LongWritable;

import org.apache.hadoop.io.Text;

//这是一个简单的MapReduce例子，进行单词数量的统计操作;

import org.apache.hadoop.mapreduce.Mapper;

/**

* KEYIN:默认情况下，是mr框架所读到的一行文本的起始偏移量，Long类型，但是在Hadoop中有更精简的序列化接口，因此采用LongWritable类型;

* VALUEIN：默认情况下，是mr框架所读到的一行文本的内容，String类型的，同上用Text(org.apache.hadoop.io.Text)类型;

* KEYOUT：是用户自定义逻辑处理完成之后输出数据中的key，在此处是单词，为String类型,同上用Text类型;

* VALUEOUT：是用户自定义逻辑处理完成之后输出数据中的value，在此处是单词数量，为Integer类型，同上用IntWritable类型;

* */

public class WordcountMapper extends Mapper

{

/**

* map阶段的业务逻辑就写在自定义的map()方法中，

* maptask会对每一行输入数据调用一次我们自定义的map()方法;

* */

@Override //覆写Mapper中的方法;

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)); //进行类型转换一下;

}无锡×××医院 https://yyk.familydoctor.com.cn/20612/

}

最后编写WordCountReduceer:

package com.jym.hadoop.mr.demo;

import java.io.IOException;

import java.util.Iterator;

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是总次数*/

public class WordcountReducerr extends Reducer

{

/**

* 例如：

*

* 输入参数key，是一组相同单词kv对的key

* */

@Override

protected void reduce(Text key, Iterable values,Context context) throws IOException, InterruptedException

{

int count= 0;

/* //采用迭代器的方式进行统计单词的数量;

Iterator iterator = values.iterator();

while(iterator.hasNext())

{

count+=iterator.next().get(); //获取key对应的value值

}

*/

//下面的for循环和上面注释中的效果是一样的;

for(IntWritable value:values)

{

count+=value.get();

}

//输出统计结果

context.write(key, new IntWritable(count));

/**

* 默认情况下reduce task会将输出结果放到一个文件中(最好是HDFS文件系统上的一个文件)

* */

}

}

然而还可以编写一个Combiner类：

package com.jym.hadoop.mr.demo;

import java.io.IOException;

import org.apache.hadoop.io.IntWritable;

import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.Reducer;

/*

* 此处的这个combiner其实不用自己编写，因为combiner的工作原理同reducer的原理是一样

* 的，故可以直接反射调用WordcountReducer类即可

* */

public class WordcountCombiner extends Reducer

{

@Override

protected void reduce(Text key, Iterable values,Context context) throws IOException, InterruptedException

{

}