示例
数据：
1949-10-01 14:21:02 34c 1949-10-02 14:01:02 36c 1950-01-01 11:21:02 32c 1950-10-01 12:21:02 37c 1951-12-01 12:21:02 23c 1951-12-02 12:21:02 45c 1951-12-03 12:21:02 50c 1951-12-23 12:21:02 33c 1950-10-02 12:21:02 41c 1950-10-03 12:21:02 27c 1951-07-01 12:21:02 45c 1951-07-02 12:21:02 46c 1951-07-03 12:21:03 47c
要求：
将每年每月中的气温排名前三的数据找出来
实现：
1.每一年用一个reduce任务处理;
2.创建自定义数据类型，存储 [年-月-日-温度];
2.自己实现排序函数 根据 [年-月-温度] 降序排列，也可以在定义数据类型中进行排序;
3.自己实现分组函数，对 [年-月] 分组，reduce的key是分组结果，根据相同的分组值，统计reduce的value值，只统计三个值就可以，因为已经实现了自排序函数。
注意点：
1.自定义数据类型的使用;
2.自定义排序类的使用;
3.自定义分组类的使用，分组类对那些数据进行分组;
4.自定义分区类，分区类与reduce job个数的关系;
示例代码：
RunJob.java
1 import org.apache.hadoop.conf.Configuration; 2 import org.apache.hadoop.fs.FileSystem; 3 import org.apache.hadoop.fs.Path; 4 import org.apache.hadoop.io.LongWritable; 5 import org.apache.hadoop.io.NullWritable; 6 import org.apache.hadoop.io.Text; 7 import org.apache.hadoop.mapreduce.Job; 8 import org.apache.hadoop.mapreduce.Mapper; 9 import org.apache.hadoop.mapreduce.Reducer; 10 import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; 11 import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; 12 13 import java.io.IOException; 14 import java.text.ParseException; 15 import java.text.SimpleDateFormat; 16 import java.util.Calendar; 17 import java.util.Date; 18 19 /** 20 * weather 统计天气信息 21 * 22 * 数据： 23 * 1999-10-01 14:21:02 34c 24 * 1999-11-02 13:01:02 30c 25 * 26 * 要求： 27 * 将每年的每月中气温排名前三的数据找出来 28 * 29 * 实现： 30 * 1.每一年用一个reduce任务处理; 31 * 2.创建自定义数据类型，存储 [年-月-日-温度]; 32 * 2.自己实现排序函数 根据 [年-月-温度] 降序排列，也可以在定义数据类型中进行排序; 33 * 3.自己实现分组函数，对 [年-月] 分组，reduce的key是分组结果，根据相同的分组值，统计reduce的value值，只统计三个值就可以，因为已经实现了自排序函数。 34 * 35 * Created by Edward on 2016/7/11. 36 */ 37 public class RunJob { 38 39 public static void main(String[] args) 40 { 41 //access hdfs's user 42 System.setProperty("HADOOP_USER_NAME","root"); 43 44 Configuration conf = new Configuration(); 45 conf.set("fs.defaultFS", "hdfs://node1:8020"); 46 47 48 try { 49 FileSystem fs = FileSystem.get(conf); 50 51 Job job = Job.getInstance(conf); 52 job.setJarByClass(RunJob.class); 53 job.setMapperClass(MyMapper.class); 54 job.setReducerClass(MyReducer.class); 55 56 //需要指定 map out 的 key 和 value 57 job.setOutputKeyClass(InfoWritable.class); 58 job.setOutputValueClass(Text.class); 59 60 //设置分区 继承 HashPartitioner 61 job.setPartitionerClass(YearPartition.class); 62 //根据年份创建指定数量的reduce task 63 job.setNumReduceTasks(3); 64 65 //设置排序 继承 WritableComparator 66 //job.setSortComparatorClass(SortComparator.class); 67 68 //设置分组 继承 WritableComparator 对reduce中的key进行分组 69 job.setGroupingComparatorClass(GroupComparator.class); 70 71 FileInputFormat.addInputPath(job, new Path("/test/weather/input")); 72 73 Path path = new Path("/test/weather/output"); 74 if(fs.exists(path))//如果目录存在，则删除目录 75 { 76 fs.delete(path,true); 77 } 78 FileOutputFormat.setOutputPath(job, path); 79 80 boolean b = job.waitForCompletion(true); 81 if(b) 82 { 83 System.out.println("OK"); 84 } 85 86 } catch (Exception e) { 87 e.printStackTrace(); 88 } 89 } 90 91 92 public static class MyMapper extends Mapper<LongWritable, Text, InfoWritable, Text > { 93 94 private static SimpleDateFormat sdf=new SimpleDateFormat("yyyy-MM-dd HH:mm:ss"); 95 96 @Override 97 protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { 98 String[] str = value.toString().split("\t"); 99 100 try { 101 Date date = sdf.parse(str[0]); 102 Calendar c = Calendar.getInstance(); 103 c.setTime(date); 104 int year = c.get(Calendar.YEAR); 105 int month = c.get(Calendar.MONTH)+1; 106 int day = c.get(Calendar.DAY_OF_MONTH); 107 108 double temperature = Double.parseDouble(str[1].substring(0,str[1].length()-1)); 109 110 InfoWritable info = new InfoWritable(); 111 info.setYear(year); 112 info.setMonth(month); 113 info.setDay(day); 114 info.setTemperature(temperature); 115 116 context.write(info, value); 117 118 } catch (ParseException e) { 119 e.printStackTrace(); 120 } 121 } 122 } 123 124 public static class MyReducer extends Reducer<InfoWritable, Text, Text, NullWritable> { 125 @Override 126 protected void reduce(InfoWritable key, Iterable<Text> values, Context context) throws IOException, InterruptedException { 127 int i=0; 128 for(Text t: values) 129 { 130 i++; 131 if(i>3) 132 break; 133 context.write(t, NullWritable.get()); 134 } 135 } 136 } 137 }
InfoWritable.java
1 import org.apache.hadoop.io.WritableComparable; 2 3 import java.io.DataInput; 4 import java.io.DataOutput; 5 import java.io.IOException; 6 7 /** 8 * 自定义数据类型 继承 WritableComparable 9 * 【年-月-日-温度】 10 * Created by Edward on 2016/7/11. 11 */ 12 public class InfoWritable implements WritableComparable<InfoWritable> { 13 14 private int year; 15 private int month; 16 private int day; 17 private double temperature; 18 19 public void setYear(int year) { 20 this.year = year; 21 } 22 23 public void setMonth(int month) { 24 this.month = month; 25 } 26 27 public void setDay(int day) { 28 this.day = day; 29 } 30 31 public void setTemperature(double temperature) { 32 this.temperature = temperature; 33 } 34 35 public int getYear() { 36 return year; 37 } 38 39 public int getMonth() { 40 return month; 41 } 42 43 public int getDay() { 44 return day; 45 } 46 47 public double getTemperature() { 48 return temperature; 49 } 50 51 /** 52 * 53 * 对象比较，对温度进行倒序排序 54 */ 55 @Override 56 public int compareTo(InfoWritable o) { 57 58 int result = Integer.compare(this.getYear(),o.getYear()); 59 if(result == 0) 60 { 61 result = Integer.compare(this.getMonth(),o.getMonth()); 62 if(result == 0) 63 { 64 return -Double.compare(this.getTemperature(), o.getTemperature()); 65 } 66 else 67 return result; 68 } 69 else 70 return result; 71 72 //return this==o?0:-1; 73 } 74 75 @Override 76 public void write(DataOutput dataOutput) throws IOException { 77 dataOutput.writeInt(this.year); 78 dataOutput.writeInt(this.month); 79 dataOutput.writeInt(this.day); 80 dataOutput.writeDouble(this.temperature); 81 } 82 83 @Override 84 public void readFields(DataInput dataInput) throws IOException { 85 this.year = dataInput.readInt(); 86 this.month = dataInput.readInt(); 87 this.day = dataInput.readInt(); 88 this.temperature = dataInput.readDouble(); 89 } 90 }
YearPartition.java
1 import org.apache.hadoop.io.Text; 2 import org.apache.hadoop.mapreduce.lib.partition.HashPartitioner; 3 4 /** 5 * 6 * 创建分区，通过key中的year来创建分区 7 * 8 * Created by Edward on 2016/7/11. 9 */ 10 public class YearPartition extends HashPartitioner <InfoWritable, Text>{ 11 @Override 12 public int getPartition(InfoWritable key, Text value, int numReduceTasks) { 13 return key.getYear()%numReduceTasks; 14 } 15 }
GroupComparator.java
1 import org.apache.hadoop.io.WritableComparable; 2 import org.apache.hadoop.io.WritableComparator; 3 4 /** 5 * 创建分组类，继承WritableComparator 6 * 【年-月】 7 * Created by Edward on 2016/7/11. 8 */ 9 public class GroupComparator extends WritableComparator { 10 11 GroupComparator() 12 { 13 super(InfoWritable.class, true); 14 } 15 16 @Override 17 public int compare(WritableComparable a, WritableComparable b) { 18 InfoWritable ia = (InfoWritable)a; 19 InfoWritable ib = (InfoWritable)b; 20 21 int result = Integer.compare(ia.getYear(),ib.getYear()); 22 if(result == 0) 23 { 24 return Integer.compare(ia.getMonth(),ib.getMonth()); 25 } 26 else 27 return result; 28 } 29 }
SortComparator.java
1 import org.apache.hadoop.io.WritableComparable; 2 import org.apache.hadoop.io.WritableComparator; 3 4 /** 5 * 排序类，继承WritableComparator 6 * 排序规则【年-月-温度】 温度降序 7 * Created by Edward on 2016/7/11. 8 */ 9 public class SortComparator extends WritableComparator { 10 11 /** 12 * 调用父类的构造函数 13 */ 14 SortComparator() 15 { 16 super(InfoWritable.class, true); 17 } 18 19 20 /** 21 * 比较两个对象的大小，使用降序排列 22 * @param a 23 * @param b 24 * @return 25 */ 26 @Override 27 public int compare(WritableComparable a, WritableComparable b) { 28 29 InfoWritable ia = (InfoWritable)a; 30 InfoWritable ib = (InfoWritable)b; 31 32 int result = Integer.compare(ia.getYear(),ib.getYear()); 33 if(result == 0) 34 { 35 result = Integer.compare(ia.getMonth(),ib.getMonth()); 36 if(result == 0) 37 { 38 return -Double.compare(ia.getTemperature(), ib.getTemperature()); 39 } 40 else 41 return result; 42 } 43 else 44 return result; 45 } 46 }
