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);
}
}