hadoop随笔二之MR-wordcount小试

基于随笔一搭建的环境，跑个wordcount任务测试。

代码如下：public class WordCount {

static final String INPUT_PATH = "hdfs://192.168.67.100:9000/input";
static final String OUT_PATH = "hdfs://192.168.67.100:9000/output";

public static void main(String[] args)
throws IOException, URISyntaxException, ClassNotFoundException, InterruptedException {
Configuration conf = new Configuration();

// conf.set("fs.default.name", "hdfs://192.168.67.100:9000");
// conf.set("hadoop.job.user", "root");
// conf.set("mapred.job.tracker", "master:9001");
final FileSystem fileSystem = FileSystem.get(new URI(INPUT_PATH), conf);
final Path outPath = new Path(OUT_PATH);
if (fileSystem.exists(outPath)) {
fileSystem.delete(outPath, true);
}
// job实例
Job job = Job.getInstance(conf, WordCount.class.getName());
// 输入路径
FileInputFormat.setInputPaths(job, INPUT_PATH);

job.setMapperClass(Mapper1.class);
job.setReducerClass(Reduce1.class);

// 降序排列
job.setSortComparatorClass(TextDescComparator.class);
job.setMapOutputKeyClass(Text.class);
job.setMapOutputValueClass(IntWritable.class);

FileOutputFormat.setOutputPath(job, outPath);
job.setJar("/usr/example/Word1.jar");
// 把job提交给JobTracker运行
job.waitForCompletion(true);

}

static IntWritable mapInt = new IntWritable(1);

public static class Mapper1 extends org.apache.hadoop.mapreduce.Mapper<Object, Text, Text, IntWritable> {

@Override
protected void map(Object key, Text value,
org.apache.hadoop.mapreduce.Mapper<Object, Text, Text, IntWritable>.Context context)
throws IOException, InterruptedException {
final String[] splited = value.toString().split("\t");
for (String word : splited) {
context.write(new Text(word), mapInt);
}

}

}

public static class Reduce1 extends Reducer<Text, IntWritable, Text, IntWritable> {

@Override
protected void reduce(Text arg0, Iterable<IntWritable> arg1,
Reducer<Text, IntWritable, Text, IntWritable>.Context context)
throws IOException, InterruptedException {
int sum = 0;
for (IntWritable iw : arg1) {
sum += iw.get();
}

context.write(arg0, new IntWritable(sum));
}

}

public static class TextDescComparator extends Text.Comparator {
public int compare(byte[] b1, int s1, int l1, byte[] b2, int s2, int l2) {
return -super.compare(b1, s1, l1, b2, s2, l2);
}
}
}
该功能是实现计算单词数，及逆序排序。
public class WordCountSort {

static final String INPUT_PATH = "hdfs://192.168.67.100:9000/sort";
static final String OUT_PATH = "hdfs://192.168.67.100:9000/output";

public static void main(String[] args)
throws IOException, URISyntaxException, ClassNotFoundException, InterruptedException {
Configuration conf = new Configuration();

final FileSystem fileSystem = FileSystem.get(new URI(INPUT_PATH), conf);
final Path outPath = new Path(OUT_PATH);
if (fileSystem.exists(outPath)) {
fileSystem.delete(outPath, true);
}
// job实例
Job job = Job.getInstance(conf, WordCountSort.class.getName());
// 输入路径
FileInputFormat.setInputPaths(job, INPUT_PATH);

job.setMapperClass(Mapper1.class);
job.setReducerClass(Reduce1.class);

job.setMapOutputKeyClass(LongWritable.class);
job.setMapOutputValueClass(Text.class);

FileOutputFormat.setOutputPath(job, outPath);
job.setJar("/usr/example/WordSort.jar");
// 把job提交给JobTracker运行
job.waitForCompletion(true);

}

public static class Mapper1 extends org.apache.hadoop.mapreduce.Mapper<LongWritable, Text, LongWritable, Text> {
private static LongWritable wordCount = new
4000
LongWritable(1);
private Text word = new Text();

@Override
protected void map(LongWritable key, Text value,
org.apache.hadoop.mapreduce.Mapper<LongWritable, Text, LongWritable, Text>.Context context)
throws IOException, InterruptedException {
// final String[] splited = value.toString().split("\t");
StringTokenizer tokenizer = new StringTokenizer(value.toString());
while (tokenizer.hasMoreTokens()) {
word = new Text(tokenizer.nextToken().trim());
wordCount = new LongWritable(Integer.valueOf(tokenizer.nextToken().trim()));
context.write(wordCount, word);// <k,v>互换
}

}
}

public static class Reduce1 extends Reducer<LongWritable, Text, Text, LongWritable> {
private Text result = new Text();

@Override
protected void reduce(LongWritable key, Iterable<Text> values,
Reducer<LongWritable, Text, Text, LongWritable>.Context context)
throws IOException, InterruptedException {
for (Text val : values) {
result.set(val.toString());
context.write(result, key);// <k,v>互换
}
}

}

}
该代码基于第一次计数后按词频排序。
结果如下：