实验315分：用MapReduce实现一个基于物品的电影推荐系统 代码用户 电影 评分101 阿凡达 6101 纵横四海 9

Mapper:

public class MovieRecommendationMapper extends Mapper<LongWritable, Text, Text, Text> {

    @Override
    public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        String[] lineValues = value.toString().split("\t");
        String user = lineValues[0];
        String movie = lineValues[1];
        String rating = lineValues[2];
        
        context.write(new Text(movie), new Text(user + "," + rating));
    }
}

Reducer:

public class MovieRecommendationReducer extends Reducer<Text, Text, Text, Text> {
    
    @Override
    public void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
        Map<String, Double> ratingMap = new HashMap<>();
        for (Text value : values) {
            String[] userRating = value.toString().split(",");
            String user = userRating[0];
            double rating = Double.parseDouble(userRating[1]);
            ratingMap.put(user, rating);
        }
        
        for (String user1 : ratingMap.keySet()) {
            for (String user2 : ratingMap.keySet()) {
                if (!user1.equals(user2)) {
                    double rating1 = ratingMap.get(user1);
                    double rating2 = ratingMap.get(user2);
                    double similarity = cosineSimilarity(ratingMap.values(), rating1, rating2);
                    if (similarity > 0.5) { // 选取相似度大于0.5的用户
                        context.write(new Text(user1), new Text(user2 + ":" + similarity));
                    }
                }
            }
        }
    }
    
    private double cosineSimilarity(Collection<Double> ratings, double rating1, double rating2) {
        double dotProduct = 0.0;
        double norm1 = 0.0;
        double norm2 = 0.0;
        for (double rating : ratings) {
            dotProduct += rating1 * rating2;
            norm1 += rating1 * rating1;
            norm2 += rating2 * rating2;
        }
        return dotProduct / (Math.sqrt(norm1) * Math.sqrt(norm2));
    }
}

Driver:

public class MovieRecommendationDriver {

    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf, "movie recommendation");

        job.setJarByClass(MovieRecommendationDriver.class);
        job.setMapperClass(MovieRecommendationMapper.class);
        job.setReducerClass(MovieRecommendationReducer.class);

        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);

        FileInputFormat.addInputPath(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));

        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }
}
``