您的位置：首页 > 其它

简单的MapReduce程序wordCount

2016-12-12 00:51 555 查看

原理：

关于原理我就不多说，网上太多了。我也是网上查的，当然这些东西我不写不代表不重要。恰恰相反原理和它的具体执行流程非常重要是基础，不然学的东西都很空。而我也是小白没有自己的看法与意见，即使写也是把别人的照搬过来，没啥意思，当然随着自己理解的加深，也会更新blog。大概如下图：

MapReduce作业包括两个进程：JobTracker，TaskTracker。JobTracker包括n个TaskTracker。

TaskTracker下包括：MapTask和ReduceTask。

环境：

hadoop-2.6.4

eclipse

步骤：

一：新建java工程，并加载jar和建类

其中WorkCount.java是我建的maperReduce类，需要引入4个包（不一定是4个也许里面有不需要的。改天我在更新。。。今天太累。。还要写blog）

二：八股文式mapreduce程序

主要就分为3大块：map， reduce， client(Main方法）

Map块：

主要是粗加工，把数据分解成一个个单词然后形成，[abc, 1],[derrt, 1],[e, 1],[222, 1],[abc, 1]…….等这种key, value类型数据。

Reduce块：

reduce块的输入是Map块的输出，主要做copy（把数据从map块拿过来）， sort， reduce（计算）。

client（主要是配置mapreduce执行的配置）。

package mr;

import java.io.IOException;
import java.util.StringTokenizer;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
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;

public class WorkCount {

// Map
static class myMapper extends Mapper<LongWritable, Text, Text, IntWritable>{

private final static IntWritable one = new IntWritable(1);
private Text word = new Text();
@Override
protected void map(LongWritable key, Text value, Context context)
throws IOException, InterruptedException {
// 获取每行数据的值
String lineValue = value.toString();

// 进行分割
StringTokenizer st = new StringTokenizer(lineValue);

// 遍历
while(st.hasMoreTokens()){
// 获取每个值
String wordValue = st.nextToken();
// 设置map输出的key值
word.set(wordValue);

// 上下文输出
context.write(word, one);
}
}
}

// reducer
static class myReducer extends Reducer<Text, IntWritable, Text, IntWritable>{

private IntWritable result = new IntWritable();
@Override
protected void reduce(Text key, Iterable<IntWritable> values,
Context context) throws IOException, InterruptedException {
// TODO Auto-generated method stub
// 用于累加
int sum = 0;
// 循环遍历Iterable
for(IntWritable value : values){
sum += value.get();
}
result.set(sum);
context.write(key, result);
}

}

// client
public static void main(String[] args) throws Exception {
// 获取配置
Configuration cf = new Configuration();
// 创建Job，设置配置和job名称
Job job = new Job(cf, "wordCount");

// 1:设置Job运行的类
job.setJarByClass(WorkCount.class);

// 2:设置map和reducer
job.setMapperClass(myMapper.class);
job.setReducerClass(myReducer.class);

// 3:设置输入文件目录和输出文件目录
FileInputFormat.addInputPath(job, new Path(args[0]));
FileOutputFormat.setOutputPath(job, new Path(args[1]));

// 4:设置输出结果的key和value类型
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(IntWritable.class);
// 5:提交Job
boolean resultB = job.waitForCompletion(true);
// 结束程序
System.exit(resultB ? 0 : 1);
}

}