hadoop电影评分排序

深藏阁楼爱情的钟 2024-04-18 15:50 1阅读 0赞

package Use.Rates;
    
    import org.apache.hadoop.io.WritableComparable;
    
    import java.io.DataInput;
    import java.io.DataOutput;
    import java.io.IOException;
    
    public class UseRatesSum  implements WritableComparable<UseRatesSum> {
        private String movie;
        private String timeStamp;
        private Integer rate;
        private String uid;
        private Integer rateSum;
    
        public String getMovie() {
            return movie;
        }
    
        public void setMovie(String movie) {
            this.movie = movie;
        }
    
        public String getTimeStamp() {
            return timeStamp;
        }
    
        public void setTimeStamp(String timeStamp) {
            this.timeStamp = timeStamp;
        }
    
        public Integer getRate() {
            return rate;
        }
    
        public void setRate(Integer rate) {
            this.rate = rate;
        }
    
        public String getUid() {
            return uid;
        }
    
        public void setUid(String uid) {
            this.uid = uid;
        }
    
        public Integer getRateSum() {
            return rateSum;
        }
    
        public void setRateSum(Integer rateSum) {
            this.rateSum = rateSum;
        }
    
        @Override
        public String toString() {
            return "movie:" + movie + ";" + "rateSum:" + rateSum;
        }
    
        @Override
        public int compareTo(UseRatesSum other) {
            Integer other_rateSum = other.getRateSum();
            Integer my_rateSum = this.rateSum;
            Integer cha = other_rateSum - my_rateSum;
            return cha;
    
        }
        //序列化
        @Override
        public void write(DataOutput dataOutput) throws IOException {
            dataOutput.writeUTF(this.movie);
            dataOutput.writeInt(this.rate);
    
        }
        //反序列化
        @Override
        public void readFields(DataInput dataInput) throws IOException {
            this.movie = dataInput.readUTF();
            this.rateSum = dataInput.readInt();
    
        }
    }

package Use.Rates;
    
    import com.doit.sumrate.UserRateDriver;
    import org.apache.hadoop.conf.Configuration;
    import org.apache.hadoop.fs.Path;
    import org.apache.hadoop.io.IntWritable;
    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;
    import org.codehaus.jackson.map.ObjectMapper;
    
    import java.io.IOException;
    
    class UserRateSumDriver {
        //定义一个map类
        public static class UserRateSumMap extends Mapper<LongWritable, Text,Text, IntWritable>{
            ObjectMapper objectMapper = new ObjectMapper();
            @Override
            protected void map(LongWritable key, Text value, Context context) throws IOException,InterruptedException{
    
                //将json数据转换成Java对象
                String line = value.toString();
                UseRatesSum useRatesSum = objectMapper.readValue(line,UseRatesSum.class);
                //获取需要的数据
                String movie = useRatesSum.getMovie();
                Integer rate = useRatesSum.getRate();
                context.write(new Text(movie),new IntWritable(rate));
    
            }
            //定义一个reduce类
        public static class UserRateSumReducer extends Reducer<Text,IntWritable,Text,IntWritable> {
            @Override
            protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
                Integer count = 0;
                for(IntWritable v : values ){
                    count = count + v.get();
                }
                context.write(key, new IntWritable(count));
            }
        }
    
    
        public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
            Configuration conf = new Configuration();
            Job job  = Job.getInstance(conf);
    
            //需要执行的jar包
            job.setJarByClass(UserRateSumDriver.class);
    
            //需要的map，reduce类
            job.setMapperClass(UserRateSumMap.class);
            job.setReducerClass(UserRateSumReducer.class);
    
            //
            job.setOutputKeyClass(Text.class);
            job.setOutputValueClass(IntWritable.class);
    
            FileInputFormat.setInputPaths(job,new Path(args[0]));
            FileOutputFormat.setOutputPath(job,new Path(args[1]));
    
            boolean b = job.waitForCompletion(true);
            System.exit(b?0:1);
    
    
    
        }
        }
    }