不多说，直接上代码。
统计出每个年龄段的 男、女 学生的最高分
这里，为了空格符的差错，直接，我们有时候，像如下这样的来排数据。
代码
package zhouls.bigdata.myMapReduce.Gender;
import java.io.IOException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Partitioner;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
/**
*
* @function 统计不同年龄段内 男、女最高分数
*
*
*/
/*
Alice<tab>23<tab>female<tab>45
Bob<tab>34<tab>male<tab>89
Chris<tab>67<tab>male<tab>97
Kristine<tab>38<tab>female<tab>53
Connor<tab>25<tab>male<tab>27
Daniel<tab>78<tab>male<tab>95
James<tab>34<tab>male<tab>79
Alex<tab>52<tab>male<tab>69
Nancy<tab>7<tab>female<tab>98
Adam<tab>9<tab>male<tab>37
Jacob<tab>7<tab>male<tab>23
Mary<tab>6<tab>female<tab>93
Clara<tab>87<tab>female<tab>72
Monica<tab>56<tab>female<tab>92
*/
public class Gender extends Configured implements Tool {
/*
*
* @function Mapper 解析输入数据，然后按需求输出
* @input key=行偏移量 value=学生数据
* @output key=gender value=name+age+score
*
*/
public static class PCMapper extends Mapper<Object, Text, Text, Text>
{
public void map(Object key, Text value, Context context) throws IOException, InterruptedException
{//拿Alice<tab>23<tab>female<tab>45
String[] tokens = value.toString().split("<tab>");//使用分隔符<tab>，将数据解析为数组 tokens
//得到Alice 23 female 45
//即tokens[0] tokens[1] tokens[2] tokens[3]
String gender = tokens[2].toString();//性别
String nameAgeScore = tokens[0] + "\t" + tokens[1] + "\t"+ tokens[3];
//输出 key=gender value=name+age+score
//输出 key=female value=Alice +23+45
context.write(new Text(gender), new Text(nameAgeScore));//将 （female ， Alice+ 23+ 45) 写入到context中
}
}
public static class MyHashPartitioner extends Partitioner<Text, Text>
{
/** Use {@link Object#hashCode()} to partition. */
@Override
public int getPartition(Text key, Text value,int numReduceTasks)
{
return (key.hashCode()) % numReduceTasks;
}
}
/**
*
* @function Partitioner 根据 age 选择 reduce 分区
*
*/
public static class PCPartitioner extends Partitioner<Text, Text>
{
@Override
public int getPartition(Text key, Text value, int numReduceTasks)
{
// TODO Auto-generated method stub
String[] nameAgeScore = value.toString().split("\t");
String age = nameAgeScore[1];//学生年龄
int ageInt = Integer.parseInt(age);//按年龄段分区
// 默认指定分区 0
if (numReduceTasks == 0)
return 0;
//年龄小于等于20，指定分区0
if (ageInt <= 20) {
return 0;
}
// 年龄大于20，小于等于50，指定分区1
if (ageInt > 20 && ageInt <= 50) {
return 1 % numReduceTasks;
}
// 剩余年龄，指定分区2
else
return 2 % numReduceTasks;
}
}
/**
*
* @function 定义Combiner 合并 Mapper 输出结果
*
*/
public static class PCCombiner extends Reducer<Text, Text, Text, Text>
{
private Text text = new Text();