JavaAPI基础操作单元测试编写并通过
This commit is contained in:
123
src/main/java/com/fjy/hadoop/mapreduce/WordCountApp.java
Normal file
123
src/main/java/com/fjy/hadoop/mapreduce/WordCountApp.java
Normal file
@@ -0,0 +1,123 @@
|
||||
package com.fjy.hadoop.mapreduce;
|
||||
|
||||
import org.apache.hadoop.conf.Configuration;
|
||||
import org.apache.hadoop.fs.Path;
|
||||
import org.apache.hadoop.io.LongWritable;
|
||||
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;
|
||||
|
||||
import java.io.IOException;
|
||||
|
||||
/**
|
||||
* 使用MapReduce开发WordCount应用
|
||||
*/
|
||||
public class WordCountApp {
|
||||
/**
|
||||
* Map:读取输入文件
|
||||
* Text:类似字符串
|
||||
*/
|
||||
public static class MyMapper extends Mapper<LongWritable, Text, Text, LongWritable> {
|
||||
|
||||
LongWritable one = new LongWritable(1);
|
||||
|
||||
/**
|
||||
* @param key 偏移量
|
||||
* @param value 每行的字符串
|
||||
* @param context 上下文
|
||||
* @throws IOException
|
||||
* @throws InterruptedException
|
||||
*/
|
||||
@Override
|
||||
protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
|
||||
/*super.map(key, value, context);*/
|
||||
|
||||
//接收到每一行数据
|
||||
String line = value.toString();
|
||||
|
||||
//按照指定分隔符进行拆分
|
||||
/*line.split("\t");//以Tab分隔*/
|
||||
String[] words = line.split(" ");//以空格分隔
|
||||
|
||||
for (String word : words) {
|
||||
//通过上下文把map的处理结果输出
|
||||
context.write(new Text(word), one);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Reduce 归并操作
|
||||
* LongWritable:文本出现的次数/求和后的次数
|
||||
*/
|
||||
public static class MyReducer extends Reducer<Text, LongWritable, Text, LongWritable> {
|
||||
/**
|
||||
* @param key
|
||||
* @param values 相同偏移量的集合
|
||||
* @param context
|
||||
* @throws IOException
|
||||
* @throws InterruptedException
|
||||
*/
|
||||
@Override
|
||||
protected void reduce(Text key, Iterable<LongWritable> values, Context context) throws IOException, InterruptedException {
|
||||
//super.reduce(key, values, context);
|
||||
|
||||
long sum = 0;
|
||||
|
||||
for (LongWritable value : values) {
|
||||
//求key出现的次数和总和
|
||||
sum += value.get();
|
||||
}
|
||||
//统计结果的输出
|
||||
context.write(key, new LongWritable(sum));
|
||||
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 定义Driver,封装MapReduce作业的所有信息
|
||||
*/
|
||||
public static void main(String[] args) throws Exception {
|
||||
|
||||
//创建Configuration
|
||||
Configuration configuration = new Configuration();
|
||||
|
||||
//创建作业
|
||||
Job job = Job.getInstance(configuration, "wordcount");
|
||||
|
||||
//设置作业的主类
|
||||
job.setJarByClass(WordCountApp.class);
|
||||
|
||||
//作业处理的输入路径
|
||||
FileInputFormat.setInputPaths(job, new Path(args[0]));
|
||||
|
||||
//设置map相关参数
|
||||
job.setMapperClass(MyMapper.class);
|
||||
|
||||
//设置map输出的key的类型
|
||||
job.setMapOutputKeyClass(Text.class);
|
||||
|
||||
//设置map输出的value的类型
|
||||
job.setMapOutputValueClass(LongWritable.class);
|
||||
|
||||
//设置Reduce相关参数
|
||||
job.setReducerClass(MyReducer.class);
|
||||
job.setOutputKeyClass(Text.class);
|
||||
job.setOutputValueClass(LongWritable.class);
|
||||
|
||||
//设置作业处理输出结果的输出路径
|
||||
FileOutputFormat.setOutputPath(job, new Path(args[1]));
|
||||
|
||||
//作业提交
|
||||
job.waitForCompletion(true);
|
||||
|
||||
//作业完成后退出
|
||||
System.exit(job.waitForCompletion(true) ? 0 : 1);
|
||||
|
||||
}
|
||||
|
||||
|
||||
}
|
||||
Reference in New Issue
Block a user