上一篇文章我们测试了hadoop自带的词频统计，本节将使用java完成自定义的词频统计。

1 系统、软件和前提约束

• 完成java访问HDFS服务
  https://www.jianshu.com/p/386cd966c04f
• 测试了hadoop自定义的词频统计
  https://www.jianshu.com/p/327d2af3903d

2 操作

• 1 在java访问HDFS的那个项目中，加入以下的测试类：

package net.wanho.mr;

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.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class WordCount {
	public static void main(String[] args) {
		try {
			//System.setProperty("hadoop.home.dir", "C:\\hadoop-2.7.2");
			Configuration conf = new Configuration();
			//conf.set("mapreduce.cluster.local.dir", "C:\\hadoop-2.7.2");
			Job job = Job.getInstance(conf);
			//指定本次mr job jar包运行主类
			job.setJarByClass(WordCount.class);
			//指定本次mr 所用的mapper combine reducer类分别是什么
			job.setMapperClass(TokenizerMapper.class);
			job.setCombinerClass(IntSumReducer.class);
			job.setReducerClass(IntSumReducer.class);
			
			//指定本次mr 最终输出的 k v类型
			job.setOutputKeyClass(Text.class);
			job.setOutputValueClass(IntWritable.class);
			//设置输入路径和输出路径，都是接受命令行参数
			FileInputFormat.addInputPath(job, args[0]);
			FileOutputFormat.setOutputPath(job, args[1]);
			//提交程序
			job.waitForCompletion(true);
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	/**
	 * reducer阶段处理的类
	 * Reducer<Text, IntWritable, Text, IntWritable>
	 * Text：就是reducer阶段输入的数据key类型，对应mapper的输出key类型 在本案例中 就是单词 Text
	 * IntWritable:就是reducer阶段输入的数据value类型，对应mapper的输出value类型 在本案例中 就是单词次数 IntWritable .
	 * Text:就是reducer阶段输出的数据key类型 在本案例中 就是单词 Text
	 * IntWritable:阶段输出的数据value类型 在本案例中 就是单词的总次数 IntWritable
	 *
	 */
	public static class IntSumReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
		private IntWritable result = new IntWritable();

		public void reduce(Text key, Iterable<IntWritable> values,
				Reducer<Text, IntWritable, Text, IntWritable>.Context context)
				throws IOException, InterruptedException {
			int sum = 0;
			for (IntWritable val : values) {
				sum += val.get();
			}
			this.result.set(sum);
			context.write(key, this.result);
		}
	}

	/**
	 * mapper阶段业务逻辑实现的类
	 * 
	 * Mapper<Object, Text, Text, IntWritable>
	 *
	 * Object：表示mapper数据输入的时候key的数据类型，在默认的读取数据组件下，叫InputFormat,它的行为是一行一行的读取待处理的数据
	 * 读取一行，返回一行给我们的mr程序，这种情况下 keyin就表示每一行的起始偏移量 因此数据类型是Long
	 *
	 * Text:表述mapper数据输入的时候value的数据类型，在默认的读取数据组件下 valuein就表示读取的这一行内容
	 * 因此数据类型是String
	 *
	 * Text 表示mapper数据输出的时候key的数据类型 在本案例当中 输出的key是单词 因此数据类型是 String
	 *
	 * IntWritable表示mapper数据输出的时候value的数据类型 在本案例当中 输出的key是单词的次数 因此数据类型是 Integer
	 * 
	 */
	public static class TokenizerMapper extends Mapper<Object, Text, Text, IntWritable> {
		private static final IntWritable one = new IntWritable(1);

		public void map(Object key, Text value, Mapper<Object, Text, Text, IntWritable>.Context context)
				throws IOException, InterruptedException {
			// 拿到一行记录
			String line = value.toString();
			// 分割这条记录
			String[] words = line.split(" ");
			for (String word : words) {
				// 使用mr程序的上下文context 把mapper阶段处理的数据发送出去,作为reduce节点的输入数据
				context.write(new Text(word), one);
			}
		}
	}
}

• 2 将项目打成jar包，作者将此jar包命名为wordcount.jar，上传到hadoop所在的linux服务器，作者将此jar包上传到了/root目录下
• 3 执行以下命令

# 切换到家目录
cd
# 进行词频统计
hadoop-2.5.2/bin/yarn jar wordcount.jar WordCount /word /output1
# 查看统计结果
hadoop-2.5.2/bin/hdfs dfs -cat /output1/part-r-00000

上面的操作就是我们通过java代码自定义词频统计的过程。

06 java完成自定义的词频统计

原文：https://www.cnblogs.com/alichengxuyuan/p/12576938.html