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

ゝ一世哀愁。 2022-06-13 05:23 127阅读 0赞

自定义类：

package myhadoop;
    
    import java.io.DataInput;
    import java.io.DataOutput;
    import java.io.IOException;
    
    import org.apache.hadoop.io.WritableComparable;
    
    public class KeyClass implements WritableComparable<KeyClass>{
    
    	private int year;
    	private int hot;
    	
    	
    	public int getYear() {
    		return year;
    	}
    
    	public void setYear(int year) {
    		this.year = year;
    	}
    
    	public int getHot() {
    		return hot;
    	}
    
    	public void setHot(int hot) {
    		this.hot = hot;
    	}
    
    	@Override
    	public void readFields(DataInput in) throws IOException {
    		this.year = in.readInt();
    		this.hot = in.readInt();
    	}
    
    	@Override
    	public void write(DataOutput out) throws IOException {
    		out.writeInt(year);
    		out.writeInt(hot);
    	}
    
    	@Override
    	public int compareTo(KeyClass o) {
    		int result = Integer.compare(this.year, o.getYear());
    		if(result!=0){
    			return result;
    		}
    		return Integer.compare(this.hot, o.hot);
    	}
    
    	@Override
    	public String toString() {
    		return year+","+hot;
    	}
    	
    	@Override
    	public int hashCode() {
    		return Integer.valueOf(year+hot).hashCode();
    	}
    	
    }

自定义排序：

package myhadoop;
    
    import org.apache.hadoop.io.WritableComparable;
    import org.apache.hadoop.io.WritableComparator;
    
    public class KeySort extends WritableComparator {
    
    	public KeySort() {
    		super(KeyClass.class,true);
    	}
    	
    	@Override
    	public int compare(WritableComparable a, WritableComparable b) {
    		KeyClass key1 = (KeyClass) a;
    		KeyClass key2 = (KeyClass) b;
    		int result = Integer.compare(key1.getYear(), key2.getYear());
    		if(result!=0){
    			return result;
    		}
    		return -Integer.compare(key1.getHot(), key2.getHot());
    	}
    	
    }

自定义分区：

package myhadoop;
    
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.Partitioner;
    
    public class KeyPart extends Partitioner<KeyClass, Text>{
    
    	@Override
    	public int getPartition(KeyClass key, Text value, int num) {
    		return (key.getYear()*127)%num;
    	}
    	
    	
    }

自定义分组：

package myhadoop;
    
    import org.apache.hadoop.io.WritableComparable;
    import org.apache.hadoop.io.WritableComparator;
    
    public class KeyGroup extends WritableComparator{
    	public KeyGroup() {
    		super(KeyClass.class,true);
    	}
    	
    	@Override
    	public int compare(WritableComparable a, WritableComparable b) {
    		KeyClass key1 = (KeyClass) a;
    		KeyClass key2 = (KeyClass) b;
    		return  Integer.compare(key1.getYear(), key2.getYear());
    	}
    }

mapreduce：

package myhadoop;
    
    import java.io.IOException;
    import java.text.SimpleDateFormat;
    import java.util.Calendar;
    import java.util.Date;
    
    import org.apache.hadoop.conf.Configuration;
    import org.apache.hadoop.fs.Path;
    import org.apache.hadoop.io.LongWritable;
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.Job;
    import org.apache.hadoop.mapreduce.Mapper;
    import org.apache.hadoop.mapreduce.Reducer;
    import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
    import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
    
    
    public class KeyMapReduce {
    	private static SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
    	static class KeyMapper extends Mapper<LongWritable, Text, KeyClass, Text>{
    
    		@Override
    		protected void map(LongWritable key, Text value, Context context)
    				throws IOException, InterruptedException {
    			String line = value.toString();
    			String[] lines = line.split("@");
    			if(lines.length==2){
    				try {
    					Date date = sdf.parse(lines[0]);
    					Calendar calendar = Calendar.getInstance();
    					calendar.setTime(date);
    					int year = calendar.get(Calendar.YEAR);
    					String hot = lines[1].substring(0, 2);
    					KeyClass keyClass = new KeyClass();
    					keyClass.setYear(year);
    					keyClass.setHot(Integer.parseInt(hot));
    					context.write(keyClass, value);
    				} catch (Exception e) {
    					// TODO Auto-generated catch block
    					e.printStackTrace();
    				}
    			}
    		}
    		
    		
    	}
    	
    	static class KeyReduce extends Reducer<KeyClass, Text, KeyClass, Text>{
    
    		@Override
    		protected void reduce(KeyClass key, Iterable<Text> value, Context context)
    				throws IOException, InterruptedException {
    			for(Text text:value){
    				context.write(key, text);
    				
    			}
    		}
    		
    	}
    	
    	public static void main(String[] args) throws Exception {
    		Configuration configuration = new Configuration();
    		Job job = Job.getInstance();
    		job.setJobName("yearAndHot");
    		job.setJarByClass(KeyMapReduce.class);
    		job.setMapperClass(KeyMapper.class);
    		job.setReducerClass(KeyReduce.class);
    		
    		job.setSortComparatorClass(KeySort.class);
    		job.setPartitionerClass(KeyPart.class);
    		job.setGroupingComparatorClass(KeyGroup.class);
    		job.setNumReduceTasks(3);
    		
    		job.setOutputKeyClass(KeyClass.class);
    		job.setOutputValueClass(Text.class);
    		FileInputFormat.addInputPath(job, new Path("/test/data.txt"));
    		FileOutputFormat.setOutputPath(job, new Path("/test/output"));
    		
    		System.exit(job.waitForCompletion(true)==true?0:1);
    	}
    }

数据：

2010-11-02 12:12:12@45℃ 
    2010-11-03 11:11:11@40℃ 
    2010-11-04 10:10:10@38℃ 
    2011-11-02 12:12:12@45℃ 
    2011-11-03 11:11:11@56℃ 
    2011-11-04 10:10:10@34℃ 
    2012-11-02 12:12:12@54℃ 
    2012-11-03 11:11:11@56℃ 
    2012-11-04 10:10:10@33℃ 
    2012-11-05 10:00:12@23℃

输出：

[hadoop@master hadoop]$ hadoop fs -ls /test/output
    Found 4 items
    -rw-r--r--   1 hadoop supergroup          0 2017-06-27 15:53 /test/output/_SUCCESS
    -rw-r--r--   1 hadoop supergroup        105 2017-06-27 15:52 /test/output/part-r-00000
    -rw-r--r--   1 hadoop supergroup        105 2017-06-27 15:52 /test/output/part-r-00001
    -rw-r--r--   1 hadoop supergroup        140 2017-06-27 15:52 /test/output/part-r-00002
    [hadoop@master hadoop]$ hadoop fs -cat /test/output/part-r-00000
    2010,45	2010-11-02 12:12:12@45℃	
    2010,40	2010-11-03 11:11:11@40℃	
    2010,38	2010-11-04 10:10:10@38℃	
    [hadoop@master hadoop]$ hadoop fs -cat /test/output/part-r-00001
    2011,56	2011-11-03 11:11:11@56℃	
    2011,45	2011-11-02 12:12:12@45℃	
    2011,34	2011-11-04 10:10:10@34℃	
    ^[[A[hadoop@master hadoop]$ hadoop fs -cat /test/output/part-r-00002
    2012,56	2012-11-03 11:11:11@56℃	
    2012,54	2012-11-02 12:12:12@54℃	
    2012,33	2012-11-04 10:10:10@33℃	
    2012,23	2012-11-05 10:00:12@23℃