本文我们介绍下使用mapreduce进行排序的实战案例。在实际的场景中，排序是一个硬性的需求，所以我们在mapreduce中需要来实现这个排序功能。
在mapreduce中，排序会发生在map阶段，因此这里的话，我们需要在map阶段进行排序，同时mapreduce是根据输出的key进行排序的，如果输出的key是intwritable的话，则会按照数字进行排序，如果输出的key是其他类型，则会按照ascii码进行排序。下面我们来演示下，这里还是使用之前的wordcount进行排序。下面直接开始
1）编写map进行words的切分
这里我们首先肯定是需要对单词进行切分，切分后的结果是：
this 1 is 1 a 1 test 1
类似上面的示例数据，详细的代码如下：
package com.mr.demo.map;
import java.io.IOException;
import java.util.StringTokenizer;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
*
* 从hdfs上读取文本内容的时候，这里使用mapper需要如下的几个输入值：
*<输入键类型，输入值类型，输出键类型，输出值类型> 这种进行类型的对标，对于<输入键类型，输入值类型>我们可以统一使用<Object, Text>即可
*
* 备注：对于上诉的<Object, Text>，仅限于text文本类型，不适合其他类型
*
*
* @author Administrator
*
*/
public class WordCountMapper extends Mapper<Object, Text, Text, IntWritable>{
private static final Logger LOGGER = LoggerFactory.getLogger(WordCountMapper.class);
private static final IntWritable one = new IntWritable(1);
private Text word = new Text();
@Override
protected void cleanup(Mapper<Object, Text, Text, IntWritable>.Context context)
throws IOException, InterruptedException {
super.cleanup(context);
}
@Override
protected void map(Object key, Text value, Mapper<Object, Text, Text, IntWritable>.Context context)
throws IOException, InterruptedException {
LOGGER.info("当前解析到的行内容是：{}",value.toString());
//这里的value就是我们接受到的每一行的内容，我们主要针对于每一行的内容进行解析
StringTokenizer words = new StringTokenizer(value.toString());
while(words.hasMoreTokens()) {
word.set(words.nextToken());
context.write(word, one);
}
//上诉我们的输出内容是： this 1,is 1,a 1,word 1 这种一个单词，数量为1
}
@Override
public void run(Mapper<Object, Text, Text, IntWritable>.Context context) throws IOException, InterruptedException {
super.run(context);
}
@Override
protected void setup(Mapper<Object, Text, Text, IntWritable>.Context context)
throws IOException, InterruptedException {
super.setup(context);
}
}2）准备reduce进行统计
在前面我们通过map把单词切分了，接下来我们就来统计下这里的单词，最后形成的数据样例如下：
this 5 is 2 a 1 test 10
这种，前面是单词，后面是单词的数量，详细的示例代码如下：
package com.mr.demo.reduce;
import java.io.IOException;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class WordCountReduce extends Reducer<Text, IntWritable, Text, IntWritable> {
private static final Logger LOGGER = LoggerFactory.getLogger(WordCountReduce.class);
private IntWritable _sum = new IntWritable();
@Override
protected void cleanup(Reducer<Text, IntWritable, Text, IntWritable>.Context context)
throws IOException, InterruptedException {
super.cleanup(context);
}
@Override
protected void reduce(Text key, Iterable<IntWritable> value,
Reducer<Text, IntWritable, Text, IntWritable>.Context context) throws IOException, InterruptedException {
// 到这里经过之前的map解析之后，然后经过shuffle传递过来的值样例是：this:[1,2,3,4,1],is:[1,2,2,1,2]
// 所以我们只需要把后面数组里面的值相加即可。
int sum = 0;
for (IntWritable val : value) {
sum += val.get();
}
_sum.set(sum);
LOGGER.info("对应的key：{} 总数量是：{}",key,sum);
context.write(key, _sum);
}
@Override
public void run(Reducer<Text, IntWritable, Text, IntWritable>.Context arg0)
throws IOException, InterruptedException {
super.run(arg0);
}
@Override
protected void setup(Reducer<Text, IntWritable, Text, IntWritable>.Context context)
throws IOException, InterruptedException {
super.setup(context);
}
}3）转换下单词与总数的个数
这里我们再使用一个map，转换下这个单词和数量的个数，原来的示例如下：
this 5
转换后的结果是：
5 this
这样子我们后续就可以根据key进行排序了，示例代码的map如下：
package com.mr.demo.map;
import java.io.IOException;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
public class SortMap extends Mapper<Object, Text, IntWritable, Text> {
private IntWritable _key = new IntWritable();
private Text _value = new Text();
@Override
protected void cleanup(Mapper<Object, Text, IntWritable, Text>.Context context)
throws IOException, InterruptedException {
super.cleanup(context);
}
@Override
protected void map(Object key, Text value, Mapper<Object, Text, IntWritable, Text>.Context context)
throws IOException, InterruptedException {
String line = value.toString().trim();
String[] tokens = line.split("\t");
_key.set(Integer.valueOf(tokens[1]));
_value.set(tokens[0]);
context.write(_key, _value);
}
@Override
public void run(Mapper<Object, Text, IntWritable, Text>.Context context) throws IOException, InterruptedException {
super.run(context);
}
@Override
protected void setup(Mapper<Object, Text, IntWritable, Text>.Context context)
throws IOException, InterruptedException {
super.setup(context);
}
}4）定义排序类型
这里的话我们可以根据key进行升序排序，也可以根据key进行倒序排序，因此这里我们需要定义一个排序类，示例代码如下：
package com.mr.demo.sort;
import org.apache.hadoop.io.IntWritable;
public class WordCountSort extends IntWritable.Comparator{
@Override
public int compare(byte[] b1, int s1, int l1, byte[] b2, int s2, int l2) {
//这里直接使用默认方法即可，如果是正数，代表升序，如果是负数，代表降序
return -super.compare(b1, s1, l1, b2, s2, l2);
}
}5）编排job
这里比较特殊，我们可以看到有2个map，1个reduce，然后整个是需要流程进行的，因此这里主类的话我们需要创建两个job，然后挨个执行，示例代码如下：
package com.mr.demo;
import java.io.IOException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
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.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import com.mr.demo.map.SortMap;
import com.mr.demo.map.WordCountMapper;
import com.mr.demo.reduce.WordCountReduce;
import com.mr.demo.sort.WordCountSort;
/**
* 这里我们首先编写一个计数的job
*
* @author Administrator
*
*/
public class WordCountJobs {
public static void main(String[] args) throws Exception {
// 在运行mapreduce的时候，我们一般都是直接执行hadoop
// -jar的时候会把这个动态的参数传递进来，因此我们还是尽量保持一致，但是这里的话，我们主要是在本地运行测试，因此这里做下判断，在本地运行的时候稍微简化一点。
if (null == args || args.length == 0) {
args = new String[] {
// 这里的输入文件可以是一个文件夹，也可以是一个具体的文件路径
"/wordcount",
//临时文件夹
"/wordcounttmp",
// 这里的输出需要指定一个文件夹
"/wordcounttarget" };
}
// 首先都要创建下这里的configuration，这里我们由于是演示的项目，因此没有特别需要配置configuration的地方，如果没有单独配置，那么所有的配置就直接拿取整个hadoop集群的配置。
Configuration configuration = new Configuration();
// 首先我们看看target目录是否存在，存在的话把他给手动删除一下
deleteOutPut(args[1], configuration);
deleteOutPut(args[2], configuration);
// 然后我们在这里编写mapreduce，来计数
Job job = Job.getInstance(configuration, "word count");
//设置运行的类，一般都是本类
job.setJarByClass(WordCountJobs.class);
//设置mapper类
job.setMapperClass(WordCountMapper.class);
//设置reduce类
job.setReducerClass(WordCountReduce.class);
//指定输出的内容key
job.setOutputKeyClass(Text.class);
//指定输出的内容value
job.setOutputValueClass(IntWritable.class);
//添加需要读取的数据源目录
FileInputFormat.addInputPath(job, new Path(args[0]));
//添加需要输出的数据目录
FileOutputFormat.setOutputPath(job, new Path(args[1]));
int code = job.waitForCompletion(true) ? 0 : 1;
if (code == 0) {
Job jobs = Job.getInstance(configuration, "word count sort");
jobs.setJarByClass(WordCountSort.class);
jobs.setMapperClass(SortMap.class);
jobs.setOutputFormatClass(TextOutputFormat.class);
jobs.setOutputKeyClass(IntWritable.class);
jobs.setOutputValueClass(Text.class);
FileInputFormat.addInputPath(jobs, new Path(args[1]));
FileOutputFormat.setOutputPath(jobs, new Path(args[2]));
jobs.setSortComparatorClass(WordCountSort.class);
code = jobs.waitForCompletion(true) ? 0 : 1;
}
System.exit(code);
}
/**
* 去服务器上删除掉输出目录，避免输出目录已经存在而报错
*
* @param args
* @throws IOException
*/
private static void deleteOutPut(String path, Configuration configuration) throws IOException {
FileSystem fileSystem = FileSystem.get(configuration);
if (fileSystem.exists(new Path(path))) {
fileSystem.delete(new Path(path), true);
}
}
}可以看到这里由2个job，每个job都有自己的输出和输出，然后第一个job执行完毕之后再会执行第二个job。
6）把项目打包上传服务器运行下：
hadoop jar MRDemo-1.0-jar-with-dependencies.jar com.mr.demo.WordCountJobs
我们可以看到在yarn上执行了两个任务：
和
最后我们看下结果输出：
可以看到实现了倒序的排序。最后按照惯例，附上本案例的源码，登录后即可下载。
还没有评论，来说两句吧...