1 协同过滤算法
协同过滤算法是现在推荐系统的一种常用算法。分为user-CF和item-CF。
本文的电影推荐系统使用的是item-CF，主要是由于用户数远远大于电影数，构建矩阵的代价更小；另外，电影推荐系统中使用基于物品的推荐对用户来说更有说服力。因此本文对user-CF只做简单介绍，主要介绍item-CF。
1.1 基于用户的协同过滤算法
a 计算出用户两两之间的相似度，得到用户相似度矩阵；
b 预测用户的喜好，使用公式：
其中，p(u,i)表示用户u对物品i的感兴趣程度，S(u,k)表示和用户u兴趣最接近的K个用户，N(i)表示对物品i有过行为的用户集合，Wuv表示用户u和用户v的兴趣相似度，Rvi表示用户v对物品i的兴趣。
c 根据预测出来的喜好度来做推荐。
1.2 基于物品的协同过滤算法
1.2.1 物品相似度计算
物品相似度的计算有多种。在这里使用同现矩阵。其中第m行第n列的元素表示物品m和物品n的相似度，具体是：如果一个用户同时看过电影m和n，则m和n的相似度就加1。还要对如下所示：
之后还要对同现矩阵做归一化，注意归一化之后矩阵不是对称的：
1.2.2 预测用户对未看电影的打分
用户打分的预测值由下式计算：
因此，最后得到的预测矩阵可由同现矩阵与评分矩阵直接相乘得到：
1.2.3 推荐
根据预测的打分，选出未看电影中的topk即生成推荐列表。
2 mapReduce工作流程
2.1 输入数据形式
表示userID, movieID，评分
2.2 总体流程
2.3 MR1
MR1负责数据预处理，将同一个user的数据merge到一起。
mapper负责拆分数据：
reducer负责合并：
2.4 MR2
MR2负责构建同现矩阵。
mapper将一个用户看过的每部电影进行两两组合发送：
reducer负责merge这些值，就得到同现矩阵的每个单元（行号：列号）：
2.5 MR3
MR3负责将同现矩阵归一化。
mapper 负责读取上一个MR产生的同现矩阵cells，然后按行号发送到reducer(由于归一化是按行的，所以这里要以行号为Key)。
reducer将得到的一行sum之后，用原来的值除以sum得到归一化的值，然后将每个单元按照列号写入HDFS（按列号写是为之后的矩阵相乘做准备）。
因此，MR3的输入输出如下：
2.6 MR4
MR4将完成矩阵小单元相乘的工作。
mapper1负责读入归一化的同现矩阵的小单元，然后按列号发送（之前已经按列号存储了，这里直接读取并发送就行）
mapper2负责读取输入的rowdata文件，即评分矩阵的每个小单元，然后按行号（movie id）发送：
在reducer中，接收到的值分别来自同现矩阵的第x列和评分矩阵的第x行。我们知道，最终生成的预测矩阵i行j列的小单元（i，j）是等于对应的同现矩阵的（i, x）乘以评分矩阵的（x, j），再对所有x求和。而这里的reducer中聚集了所有x值相同的来自两个矩阵的小单元，因此它们两两之间是可以互乘的。这里我们用=和：来区分两个矩阵的小单元。下图中橘黄色是处于同一个reducer里面的小单元，将来自同现矩阵和评分矩阵的小单元区分开后，将它们两两相乘，得到预测矩阵的行号与列号的不同组合，以它为key写入hdfs。
2.7 MR5
MR5负责将乘积的结果相加。
3 主要代码
DataDividerByUser.java
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.TextInputFormat; import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat; import java.io.IOException; public class DataDividerByUser { public static class DataDividerMapper extends Mapper<LongWritable, Text, IntWritable, Text> { // map method @Override public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { //input user,movie,rating String[] user_movie_rating = value.toString().split(","); int userId = Integer.parseInt(user_movie_rating[0]); String outPutKey = user_movie_rating[1] + ":" + user_movie_rating[2]; //divide data by user context.write(new IntWritable(userId), new Text(outPutKey)); } } public static class DataDividerReducer extends Reducer<IntWritable, Text, IntWritable, Text> { // reduce method @Override public void reduce(IntWritable key, Iterable<Text> values, Context context) throws IOException, InterruptedException { StringBuilder sb = new StringBuilder(); //merge data for one user for (Text value : values) { sb.append(value.toString()); sb.append(","); } sb.deleteCharAt(sb.length() - 1); context.write(key, new Text(sb.toString())); } } public static void main(String[] args) throws Exception { Configuration conf = new Configuration(); Job job = Job.getInstance(conf); job.setMapperClass(DataDividerMapper.class); job.setReducerClass(DataDividerReducer.class); job.setJarByClass(DataDividerByUser.class); job.setInputFormatClass(TextInputFormat.class); job.setOutputFormatClass(TextOutputFormat.class); job.setOutputKeyClass(IntWritable.class); job.setOutputValueClass(Text.class); TextInputFormat.setInputPaths(job, new Path(args[0])); TextOutputFormat.setOutputPath(job, new Path(args[1])); job.waitForCompletion(true); } }
CoOccurrenceMatrixGenerator.java
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.TextInputFormat; import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat; import java.io.IOException; import java.util.ArrayList; import java.util.List; public class CoOccurrenceMatrixGenerator { public static class MatrixGeneratorMapper extends Mapper<LongWritable, Text, Text, IntWritable> { // map method @Override public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { //value = userid \t movie1: rating, movie2: rating... String[] movie_rating = value.toString().split("\t")[1].split(","); //key = movie1: movie2 value = 1 //calculate each user rating list: <movieA, movieB> for (int i = 0; i < movie_rating.length; i++) { for (int j = 0; j < movie_rating.length; j++) { String outPutKey = movie_rating[i].split(":")[0] + ":" + movie_rating[j].split(":")[0]; context.write(new Text(outPutKey), new IntWritable(1)); } } } } public static class MatrixGeneratorReducer extends Reducer<Text, IntWritable, Text, IntWritable> { // reduce method @Override public void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException { //key movie1:movie2 value = iterable<1, 1, 1> //calculate each two movies have been watched by how many people int sum = 0; for (IntWritable value : values) { sum += value.get(); } context.write(key, new IntWritable(sum)); } } public static void main(String[] args) throws Exception{ Configuration conf = new Configuration(); Job job = Job.getInstance(conf); job.setMapperClass(MatrixGeneratorMapper.class); job.setReducerClass(MatrixGeneratorReducer.class); job.setJarByClass(CoOccurrenceMatrixGenerator.class); job.setInputFormatClass(TextInputFormat.class); job.setOutputFormatClass(TextOutputFormat.class); job.setOutputKeyClass(Text.class); job.setOutputValueClass(IntWritable.class); TextInputFormat.setInputPaths(job, new Path(args[0])); TextOutputFormat.setOutputPath(job, new Path(args[1])); job.waitForCompletion(true); } }
Normalize.java
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.TextInputFormat; import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat; import java.io.IOException; import java.util.HashMap; import java.util.Map; public class Normalize { public static class NormalizeMapper extends Mapper<LongWritable, Text, Text, Text> { // map method @Override public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { //movieA:movieB \t relation String movieA = value.toString().split("\t")[0].split(":")[0]; String movieB = value.toString().split("\t")[0].split(":")[1]; String relation = value.toString().split("\t")[1]; //collect the relationship list for movieA context.write(new Text(movieA), new Text(movieB + ":" + relation)); } } public static class NormalizeReducer extends Reducer<Text, Text, Text, Text> { // reduce method @Override public void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException { //key = movieA, value=<movieB:relation, movieC:relation...> //normalize each unit of co-occurrence matrix Map<String, Double> map = new HashMap<String, Double>(); double sum = 0; for (Text value : values) { String[] movie_relation = value.toString().split(":"); map.put(movie_relation[0], Double.parseDouble(movie_relation[1])); sum += Double.parseDouble(movie_relation[1]); } for (Map.Entry<String, Double> entry : map.entrySet()) { String outputKey = entry.getKey(); String outputValue = key.toString() + "=" + String.valueOf(entry.getValue() / sum); context.write(new Text(outputKey), new Text(outputValue)); } } } public static void main(String[] args) throws Exception { Configuration conf = new Configuration(); Job job = Job.getInstance(conf); job.setMapperClass(NormalizeMapper.class); job.setReducerClass(NormalizeReducer.class); job.setJarByClass(Normalize.class); job.setInputFormatClass(TextInputFormat.class); job.setOutputFormatClass(TextOutputFormat.class); job.setOutputKeyClass(Text.class); job.setOutputValueClass(Text.class); TextInputFormat.setInputPaths(job, new Path(args[0])); TextOutputFormat.setOutputPath(job, new Path(args[1])); job.waitForCompletion(true); } }
Multiplication.java
import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.DoubleWritable; 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.chain.ChainMapper; import org.apache.hadoop.mapreduce.lib.input.MultipleInputs; import org.apache.hadoop.mapreduce.lib.input.TextInputFormat; import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat; import java.io.IOException; import java.util.HashMap; import java.util.List; import java.util.Map; public class Multiplication { public static class CooccurrenceMapper extends Mapper<LongWritable, Text, Text, Text> { // map method @Override public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { //input: movieB \t movieA=relation //pass data to reducer String[] movieB_movieARelation = value.toString().split("\t"); context.write(new Text(movieB_movieARelation[0]), new Text(movieB_movieARelation[1])); } } public static class RatingMapper extends Mapper<LongWritable, Text, Text, Text> { // map method @Override public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { //input: user,movie,rating //pass data to reducer String[] user_movie_rating = value.toString().split(","); String outputKey = user_movie_rating[0] + ":" + user_movie_rating[2]; context.write(new Text(user_movie_rating[1]), new Text(outputKey)); } } public static class MultiplicationReducer extends Reducer<Text, Text, Text, DoubleWritable> { // reduce method @Override public void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException { //key = movieB value = <movieA=relation, movieC=relation... userA:rating, userB:rating...> //collect the data for each movie, then do the multiplication Map<String, Double> coMap = new HashMap<String, Double>(); Map<String, Double> ratingMap = new HashMap<String, Double>(); for (Text value : values) { String s = value.toString(); if (s.contains("=")) { coMap.put(s.split("=")[0], Double.parseDouble(s.split("=")[1])); } else { ratingMap.put(s.split(":")[0], Double.parseDouble(s.split(":")[1])); } } for (Map.Entry<String, Double> entry1 : coMap.entrySet()) { for (Map.Entry<String, Double> entry2 : ratingMap.entrySet()) { double mult = entry1.getValue() * entry2.getValue(); String outputKey = entry2.getKey() + ":" + entry1.getKey(); context.write(new Text(outputKey), new DoubleWritable(mult)); } } } } public static void main(String[] args) throws Exception { Configuration conf = new Configuration(); Job job = Job.getInstance(conf); job.setJarByClass(Multiplication.class); ChainMapper.addMapper(job, CooccurrenceMapper.class, LongWritable.class, Text.class, Text.class, Text.class, conf); ChainMapper.addMapper(job, RatingMapper.class, Text.class, Text.class, Text.class, Text.class, conf); job.setMapperClass(CooccurrenceMapper.class); job.setMapperClass(RatingMapper.class); job.setReducerClass(MultiplicationReducer.class); job.setMapOutputKeyClass(Text.class); job.setMapOutputValueClass(Text.class); job.setOutputKeyClass(Text.class); job.setOutputValueClass(DoubleWritable.class); MultipleInputs.addInputPath(job, new Path(args[0]), TextInputFormat.class, CooccurrenceMapper.class); MultipleInputs.addInputPath(job, new Path(args[1]), TextInputFormat.class, RatingMapper.class); TextOutputFormat.setOutputPath(job, new Path(args[2])); job.waitForCompletion(true); } }
Sum.java
import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.DoubleWritable; 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.TextInputFormat; import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat; import java.io.IOException; /** * Created by Michelle on 11/12/16. */ public class Sum { public static class SumMapper extends Mapper<LongWritable, Text, Text, DoubleWritable> { // map method @Override public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { //pass data to reducer String[] key_value = value.toString().split("\t"); context.write(new Text(key_value[0]), new DoubleWritable(Double.parseDouble(key_value[1]))); } } public static class SumReducer extends Reducer<Text, DoubleWritable, Text, DoubleWritable> { // reduce method @Override public void reduce(Text key, Iterable<DoubleWritable> values, Context context) throws IOException, InterruptedException { //user:movie relation //calculate the sum double sum = 0; for (DoubleWritable value : values) { sum += value.get(); } context.write(key, new DoubleWritable(sum)); } } public static void main(String[] args) throws Exception { Configuration conf = new Configuration(); Job job = Job.getInstance(conf); job.setMapperClass(SumMapper.class); job.setReducerClass(SumReducer.class); job.setJarByClass(Sum.class); job.setInputFormatClass(TextInputFormat.class); job.setOutputFormatClass(TextOutputFormat.class); job.setOutputKeyClass(Text.class); job.setOutputValueClass(DoubleWritable.class); TextInputFormat.setInputPaths(job, new Path(args[0])); TextOutputFormat.setOutputPath(job, new Path(args[1])); job.waitForCompletion(true); } }
Driver.java
public class Driver { public static void main(String[] args) throws Exception { DataDividerByUser dataDividerByUser = new DataDividerByUser(); CoOccurrenceMatrixGenerator coOccurrenceMatrixGenerator = new CoOccurrenceMatrixGenerator(); Normalize normalize = new Normalize(); Multiplication multiplication = new Multiplication(); Sum sum = new Sum(); String rawInput = args[0]; String userMovieListOutputDir = args[1]; String coOccurrenceMatrixDir = args[2]; String normalizeDir = args[3]; String multiplicationDir = args[4]; String sumDir = args[5]; String[] path1 = {rawInput, userMovieListOutputDir}; String[] path2 = {userMovieListOutputDir, coOccurrenceMatrixDir}; String[] path3 = {coOccurrenceMatrixDir, normalizeDir}; String[] path4 = {normalizeDir, rawInput, multiplicationDir}; String[] path5 = {multiplicationDir, sumDir}; dataDividerByUser.main(path1); coOccurrenceMatrixGenerator.main(path2); normalize.main(path3); multiplication.main(path4); sum.main(path5); } }
