mapreduce，自定义排序，分区，分组实现按照年份升序排序，温度降序排序

末蓝、 2022-05-20 10:28 183阅读 0赞

自定义类：

1.  `package myhadoop;`
2.   
3.  `import java.io.DataInput;`
4.  `import java.io.DataOutput;`
5.  `import java.io.IOException;`
6.   
7.  `import org.apache.hadoop.io.WritableComparable;`
8.   
9.  `public class KeyClass implements WritableComparable<KeyClass>{`
10.  
11. `private int year;`
12. `private int hot;`
13.  
14.  
15. `public int getYear() {`
16. `return year;`
17. `}`
18.  
19. `public void setYear(int year) {`
20. `this.year = year;`
21. `}`
22.  
23. `public int getHot() {`
24. `return hot;`
25. `}`
26.  
27. `public void setHot(int hot) {`
28. `this.hot = hot;`
29. `}`
30.  
31. `@Override`
32. `public void readFields(DataInput in) throws IOException {`
33. `this.year = in.readInt();`
34. `this.hot = in.readInt();`
35. `}`
36.  
37. `@Override`
38. `public void write(DataOutput out) throws IOException {`
39. `out.writeInt(year);`
40. `out.writeInt(hot);`
41. `}`
42.  
43. `@Override`
44. `public int compareTo(KeyClass o) {`
45. `int result = Integer.compare(this.year, o.getYear());`
46. `if(result!=0){`
47. `return result;`
48. `}`
49. `return Integer.compare(this.hot, o.hot);`
50. `}`
51.  
52. `@Override`
53. `public String toString() {`
54. `return year+","+hot;`
55. `}`
56.  
57. `@Override`
58. `public int hashCode() {`
59. `return Integer.valueOf(year+hot).hashCode();`
60. `}`
61.  
62. `}`

自定义排序：

1.  `package myhadoop;`
2.   
3.  `import org.apache.hadoop.io.WritableComparable;`
4.  `import org.apache.hadoop.io.WritableComparator;`
5.   
6.  `public class KeySort extends WritableComparator {`
7.   
8.  `public KeySort() {`
9.  `super(KeyClass.class,true);`
10. `}`
11.  
12. `@Override`
13. `public int compare(WritableComparable a, WritableComparable b) {`
14. `KeyClass key1 = (KeyClass) a;`
15. `KeyClass key2 = (KeyClass) b;`
16. `int result = Integer.compare(key1.getYear(), key2.getYear());`
17. `if(result!=0){`
18. `return result;`
19. `}`
20. `return -Integer.compare(key1.getHot(), key2.getHot());`
21. `}`
22.  
23. `}`

自定义分区：

1.  `package myhadoop;`
2.   
3.  `import org.apache.hadoop.io.Text;`
4.  `import org.apache.hadoop.mapreduce.Partitioner;`
5.   
6.  `public class KeyPart extends Partitioner<KeyClass, Text>{`
7.   
8.  `@Override`
9.  `public int getPartition(KeyClass key, Text value, int num) {`
10. `return (key.getYear()*127)%num;`
11. `}`
12.  
13.  
14. `}`

自定义分组：

1.  `package myhadoop;`
2.   
3.  `import org.apache.hadoop.io.WritableComparable;`
4.  `import org.apache.hadoop.io.WritableComparator;`
5.   
6.  `public class KeyGroup extends WritableComparator{`
7.  `public KeyGroup() {`
8.  `super(KeyClass.class,true);`
9.  `}`
10.  
11. `@Override`
12. `public int compare(WritableComparable a, WritableComparable b) {`
13. `KeyClass key1 = (KeyClass) a;`
14. `KeyClass key2 = (KeyClass) b;`
15. `return Integer.compare(key1.getYear(), key2.getYear());`
16. `}`
17. `}`

mapreduce：

1.  `package myhadoop;`
2.   
3.  `import java.io.IOException;`
4.  `import java.text.SimpleDateFormat;`
5.  `import java.util.Calendar;`
6.  `import java.util.Date;`
7.   
8.  `import org.apache.hadoop.conf.Configuration;`
9.  `import org.apache.hadoop.fs.Path;`
10. `import org.apache.hadoop.io.LongWritable;`
11. `import org.apache.hadoop.io.Text;`
12. `import org.apache.hadoop.mapreduce.Job;`
13. `import org.apache.hadoop.mapreduce.Mapper;`
14. `import org.apache.hadoop.mapreduce.Reducer;`
15. `import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;`
16. `import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;`
17.  
18.  
19. `public class KeyMapReduce {`
20. `private static SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");`
21. `static class KeyMapper extends Mapper<LongWritable, Text, KeyClass, Text>{`
22.  
23. `@Override`
24. `protected void map(LongWritable key, Text value, Context context)`
25. `throws IOException, InterruptedException {`
26. `String line = value.toString();`
27. `String[] lines = line.split("@");`
28. `if(lines.length==2){`
29. `try {`
30. `Date date = sdf.parse(lines[0]);`
31. `Calendar calendar = Calendar.getInstance();`
32. `calendar.setTime(date);`
33. `int year = calendar.get(Calendar.YEAR);`
34. `String hot = lines[1].substring(0, 2);`
35. `KeyClass keyClass = new KeyClass();`
36. `keyClass.setYear(year);`
37. `keyClass.setHot(Integer.parseInt(hot));`
38. `context.write(keyClass, value);`
39. `} catch (Exception e) {`
40. `// TODO Auto-generated catch block`
41. `e.printStackTrace();`
42. `}`
43. `}`
44. `}`
45.  
46.  
47. `}`
48.  
49. `static class KeyReduce extends Reducer<KeyClass, Text, KeyClass, Text>{`
50.  
51. `@Override`
52. `protected void reduce(KeyClass key, Iterable<Text> value, Context context)`
53. `throws IOException, InterruptedException {`
54. `for(Text text:value){`
55. `context.write(key, text);`
56.  
57. `}`
58. `}`
59.  
60. `}`
61.  
62. `public static void main(String[] args) throws Exception {`
63. `Configuration configuration = new Configuration();`
64. `Job job = Job.getInstance();`
65. `job.setJobName("yearAndHot");`
66. `job.setJarByClass(KeyMapReduce.class);`
67. `job.setMapperClass(KeyMapper.class);`
68. `job.setReducerClass(KeyReduce.class);`
69.  
70. `job.setSortComparatorClass(KeySort.class);`
71. `job.setPartitionerClass(KeyPart.class);`
72. `job.setGroupingComparatorClass(KeyGroup.class);`
73. `job.setNumReduceTasks(3);`
74.  
75. `job.setOutputKeyClass(KeyClass.class);`
76. `job.setOutputValueClass(Text.class);`
77. `FileInputFormat.addInputPath(job, new Path("/test/data.txt"));`
78. `FileOutputFormat.setOutputPath(job, new Path("/test/output"));`
79.  
80. `System.exit(job.waitForCompletion(true)==true?0:1);`
81. `}`
82. `}`

数据：

1.  `2010-11-02 12:12:12@45℃`
2.  `2010-11-03 11:11:11@40℃`
3.  `2010-11-04 10:10:10@38℃`
4.  `2011-11-02 12:12:12@45℃`
5.  `2011-11-03 11:11:11@56℃`
6.  `2011-11-04 10:10:10@34℃`
7.  `2012-11-02 12:12:12@54℃`
8.  `2012-11-03 11:11:11@56℃`
9.  `2012-11-04 10:10:10@33℃`
10. `2012-11-05 10:00:12@23℃`

输出：

1.  `[hadoop@master hadoop]$ hadoop fs -ls /test/output`
2.  `Found 4 items`
3.  `-rw-r--r-- 1 hadoop supergroup 0 2017-06-27 15:53 /test/output/_SUCCESS`
4.  `-rw-r--r-- 1 hadoop supergroup 105 2017-06-27 15:52 /test/output/part-r-00000`
5.  `-rw-r--r-- 1 hadoop supergroup 105 2017-06-27 15:52 /test/output/part-r-00001`
6.  `-rw-r--r-- 1 hadoop supergroup 140 2017-06-27 15:52 /test/output/part-r-00002`
7.  `[hadoop@master hadoop]$ hadoop fs -cat /test/output/part-r-00000`
8.  `2010,45 2010-11-02 12:12:12@45℃`
9.  `2010,40 2010-11-03 11:11:11@40℃`
10. `2010,38 2010-11-04 10:10:10@38℃`
11. `[hadoop@master hadoop]$ hadoop fs -cat /test/output/part-r-00001`
12. `2011,56 2011-11-03 11:11:11@56℃`
13. `2011,45 2011-11-02 12:12:12@45℃`
14. `2011,34 2011-11-04 10:10:10@34℃`
15. `^[[A[hadoop@master hadoop]$ hadoop fs -cat /test/output/part-r-00002`
16. `2012,56 2012-11-03 11:11:11@56℃`
17. `2012,54 2012-11-02 12:12:12@54℃`
18. `2012,33 2012-11-04 10:10:10@33℃`
19. `2012,23 2012-11-05 10:00:12@23℃`