协同过滤算法一般分为两种实现：
第一种基于用户的协同过滤算法适用于公司初期，用户数量不多的情况下；这里物品推荐实现采用的是第二种，基于物品的协同过滤算法。
1、用户对物品的推荐列表（用户为列，物品为行） = 用户对物品的评分矩阵 × 物品同现矩阵
2、用户对物品的评分矩阵：用户对物品的点击、收藏、加购物车和购买等行为都是对物品的不同评分，矩阵如下：
并且，矩阵可以进行行列转换
3、物品同现矩阵：两个物品出现在同一个用户的次数，即item1和item2都出现在user1和user3，所以item1:item2的同现值为2，当然item2:item1的同现值也为2，所以物品同现矩阵是一个对称矩阵，对称矩阵的行列转换不会改变数据形态，如下：
4、矩阵相乘（行*列）计算可以得到推荐列表：
可以看到，两种计算方法得到的结果都是一样的，只是矩阵展示形式不一样，user1的优先推荐物品都是item2，评分为9，其次是item1，评分是8，最后是item3，评分是5。
一、使用itemId作为key，来进行矩阵的行列计算，因为：
1、reduce一次只计算一组相同key的数据；
2、行列计算需要用户评分矩阵数据和物品同现矩阵树；
3、物品同现矩阵的行和列都是itemId，所以只能用itemId。
二、因为是以itemId作为key来计算，所以要给用户对物品评分矩阵做行列转换，如上A行列转换变成B，这样在进行redice计算是就可以输入所有用户对物品itemId的评分。
三、MapReduce使用的计算方式应该是第一种 物品对用户的评分矩阵A * 物品同现矩阵C
四、行列计算分析，等号前面是行列计算各乘积算和，就可以得到对应用户对物品的推荐值：
user1:item1 * item1:item1 + user1:item2 * item2:item1 + user1:item3 * item3:item1 = user1:item1
user1:item1 * item1:item2 + user1:item2 * item2:item2 + user1:item3 * item3:item2 = user1:item2
user1:item1 * item1:item3 + user1:item2 * item2:item3 + user1:item3 * item3:item3 = user1:item3
注意，这里每一个乘积的因子都有一个共同的itemId，且用户对物品的推荐值与乘积因子的其他两个元素相对应，如user1:item1，因此矩阵相乘计算可以分以下两步reduce完成：
1、第一次reduce对同一物品id计算所有的 评分 * 同现值，获得一个乘积列表，包括如：user1:item1 * item1:item1、user1:item1 * item1:item2，输出已userId为key，item1、item2拼上乘积作为key为vaue；
2、第二次reduce求和，以user1:item1为一类，计算user1对item1的推荐总值。所有用户对所有物品都计算求和，就可以得到用户对物品的推荐列表。
user2,item3,click
user1,item1,click
user1,item1,collect
user1,item2,click
user2,item2,collect
user2,item3,click
user3,item1,collect
user3,item2,click
user1,item2,click
user3,item1,collect
user1,item2,click
user3,item3,click如果原始数据记录有重复，则需要一个MapReduce进行去重，即用户对物品的同一行为的多条数据去重。可以设置输入格式化类为KeyValueTextInputFormat，然后Map只做透传，可以直接用框架的Mapper类，reduce只输出第一条记录，主要代码如下：
//设置输入格式化类，是的整条记录作为key
job.setInputFormatClass(KeyValueTextInputFormat.class);public class ItemCFStep1Reducer extends Reducer<Text, NullWritable, Text, NullWritable> {
@Override
protected void reduce(Text key, Iterable<NullWritable> values, Context context) throws IOException, InterruptedException {
//只输出一个
context.write(key, NullWritable.get());
}
}一、mpper的输入数据是上一步的结果集，输出用户对应物品的评分记录，如user1 item1:1。代码如下：
public class ItemCFStep2Mapper extends Mapper<Text, Text, Text, Text> {
private final Text mkey = new Text();
private final Text mval = new Text();
@Override
protected void map(Text key, Text value, Context context) throws IOException, InterruptedException {
//样本数据：key i1890 value u10218,collect
StringTokenizer st = new StringTokenizer(value.toString(), ",");
String userId = st.nextToken();
//用户对物品评分
int recode = UserAction.getRecord(st.nextToken());
mkey.set(userId);
mval.set(key.toString() + ":" + recode);
context.write(mkey, mval);
}
}注意，这里创建job时需要先设置输入格式化类，以及分割符，如下：
job.getConfiguration().set("mapreduce.input.keyvaluelinerecordreader.key.value.separator", ",");
job.setInputFormatClass(KeyValueTextInputFormat.class);二、reducer需要完成下面两步工作：
1、统计各用户对个物品的评分； 2、合并同个用户的所有物品及其评分；
public class ItemCFStep2Reducer extends Reducer<Text, Text, Text, Text> {
@Override
protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
//样本数据：
//u10218 i1890:1
HashMap<String, Integer> map = new HashMap<>();
for (Text value : values) {
//对相同用户的相同商品累计分数
StringTokenizer st = new StringTokenizer(value.toString(), ":");
String itemId = st.nextToken();
int tmpRecord = Integer.parseInt(st.nextToken());
Integer record = map.get(itemId);
map.put(itemId, record == null ? tmpRecord : tmpRecord + record);
}
StringBuffer sb = new StringBuffer();
for (Map.Entry<String, Integer> entry : map.entrySet()) {
sb.append(entry.getKey()).append(":").append(entry.getValue()).append(",");
}
//输出的同一个用户的物品不会重复
context.write(key, new Text(sb.substring(0, sb.length() - 1)));
}
}三、最终获得的结果集如下：
user1 item1:3,item2:1
user2 item2:2,item3:1
user3 item1:2,item2:1,item3:1一、mapper的输入数据是上一步的结果集，然对各个用户的物品评分列表里的物品两两输出，一条记录计数为1。注意，物品自己对自己也要输出一条记录。
public class ItemCFStep3Mapper extends Mapper<Text, Text, Text, IntWritable> {
private final Text mkey = new Text();
private final IntWritable mval = new IntWritable(1);
@Override
protected void map(Text key, Text value, Context context) throws IOException, InterruptedException {
//数据样本：u10224 i1500:3,i1748:2,i1627:4,i1966:3
String[] userReocers = StringUtils.split(value.toString(), ',');
for (int i = 0; i < userReocers.length; i++) {
String item1 = userReocers[i].split(":")[0];
//输出自己
mkey.set(item1 + ":" + item1);
context.write(mkey, mval);
for (int j = i + 1; j < userReocers.length; j++) {
mkey.set(new Text(item1 + ":" + userReocers[j].split(":")[0]));
context.write(mkey, mval);
}
}
}
}二、reducer对相同的两个物品统计同现次数，注意！！！输出时还要多输出一条镜像记录，如记录item1:item2还要输出一条item2:item1，不然得到的同现矩阵只有一半的数据。
public class ItemCFStep3Reducer extends Reducer<Text, IntWritable, Text, IntWritable> {
private final Text rkey = new Text();
private final IntWritable rval = new IntWritable();
@Override
protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
//样本数据：item1:item2 1
int count = 0;
for (IntWritable value : values) {
count += value.get();
}
rval.set(count);
String[] ss = StringUtils.split(key.toString(), ':');
if (!ss[0].equals(ss[1])) {
//注意！！！这里要对物品同现值做一个镜像输出（物品对自己的不需要），不然得到的物品同现矩阵只有一半的数据
rkey.set(ss[1] + ":" + ss[0]);
context.write(rkey, rval);
}
context.write(key,rval);
}
}三、输出结果集
item1:item1 2
item2:item1 2
item1:item2 2
item3:item1 1
item1:item3 1
item2:item2 3
item3:item2 2
item2:item3 2
item3:item3 2一、这里mapper需要处理两个数据集，第二步得到的用户评分矩阵和第三步得到的同现矩阵。两个数据集的输出都以itemId做为key，第一个数据集以userId:3作为value，第二个数据集以itemId:2作为value，所以为了在reduce中可以区分这两种数据，输入时要改value加上相应的标识。
public class ItemCFStep4Mapper extends Mapper<Text, Text, Text, Text> {
private String file = "";
private final Text mkey = new Text();
private final Text mval = new Text();
@Override
protected void setup(Context context) throws IOException, InterruptedException {
FileSplit fs = (FileSplit) context.getInputSplit();
file = fs.getPath().getParent().getName();
}
@Override
protected void map(Text key, Text value, Context context) throws IOException, InterruptedException {
//两个数据集样本
//i1000:i1000 40
//u10224 i1500:3,i1748:2,i1627:4,i1966:3
if (file.equals("output3")) {
String[] items = StringUtils.split(key.toString(), ':');
mkey.set(items[0]);
mval.set("A:" + items[1] + "," + value.toString());
context.write(mkey, mval);
} else {
String[] itemRecords = StringUtils.split(value.toString(), ',');
for (String itemRecord : itemRecords) {
String[] ss = StringUtils.split(itemRecord, ':');
mkey.set(ss[0]);
mval.set("B:" + key.toString() + "," + ss[1]);
context.write(mkey, mval);
}
}
}
}二、reduce以itemId为维度，对所有用户对该物品的评分和所有物品与该物品的同现值两两相乘，并将乘积一一输出，userId作为key，物品的itemId（注意不是作为reduce输入key的itemId）拼上乘积作为value。这样我们计算矩阵行列相乘里需要累加的乘积的数据集就得到了。主要代码逻辑：
1、mapA保存所有物品和该物品同现数次，mapB保存所有用户对该物品的评分
2、对mapA和mapB做嵌套迭代，每次内层迭代输出键值对，key为用户id，value为物品i拼上两项乘积。
两项乘积 = i物品对item1的同现次数 * j用户对item1的评分
public class ItemCFStep4Reducer extends Reducer<Text, Text, Text, Text> {
/**正则表达式可以预编译，所以最好现在外面创建表达式对象*/
public static final Pattern PATTERN = Pattern.compile("[:,]");
private final Text rkey = new Text();
private final Text rval = new Text();
@Override
protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
//样本数据：
//物品同现矩阵数据
//item1 A:item2,1
//item1 A:item1,1
//物品对用户评分数据
//item1 B:user1,2
//item1 B:user2,1
//保存同现矩阵数据，item1 1
Map<String, Integer> mapA = new HashMap<>();
//保存用户评分数据，user1 1
Map<String, Integer> mapB = new HashMap<>();
for (Text value : values) {
String[] ss = PATTERN.split(value.toString());
if ("A".equals(ss[0])) {
mapA.put(ss[1], Integer.parseInt(ss[2]));
} else if ("B".equals(ss[0])) {
mapB.put(ss[1], Integer.parseInt(ss[2]));
}
}
for (Map.Entry<String, Integer> entryA : mapA.entrySet()) {
for (Map.Entry<String, Integer> entryB : mapB.entrySet()) {
rkey.set(entryB.getKey());
rval.set(entryA.getKey() + "," + entryA.getValue() * entryB.getValue());
context.write(rkey, rval);
}
}
}
}三、输出结果集
user1 item2,6
user3 item2,4
user1 item1,6
user3 item1,4
user1 item3,3
user3 item3,2
user1 item2,3
user2 item2,6
user3 item2,3
user1 item3,2
user2 item3,4
user3 item3,2
user2 item3,2
user3 item3,2
一、mapper的输入数据是上一步的结果集，直接输出key和value，可以用默认的Mapper。
二、reduce根据用户id计算对各物品的评分，同一物品需要累计求和，即根据userId和ItemId确定一组求和，即可得到用户对某物品的推荐值。
public class ItemCFStep5Reducer extends Reducer<Text, Text, Text, Text> {
private final Text rval = new Text();
@Override
protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
//样本数据
//user1 item2,6
//user1 item1,6
//user1 item2,3
Map<String, Integer> map = new HashMap<>();
for (Text text : values) {
String[] ss = StringUtils.split(text.toString(), ',');
Integer record = map.get(ss[0]);
map.put(ss[0], record == null ? Integer.parseInt(ss[1]) : record + Integer.parseInt(ss[1]));
}
// StringBuffer sb = new StringBuffer();
for (Map.Entry<String, Integer> entry : map.entrySet()) {
rval.set(entry.getKey() + ":" + entry.getValue());
context.write(key, rval);
// sb.append(entry.getKey()).append(":").append(entry.getValue()).append(",");
}
// rval.set(sb.substring(0, sb.length() - 1));
}
}三、输出结果集：
user1:item2 9
user1:item1 8
user1:item3 5
user2:item1 5
user2:item2 8
user2:item3 6
user3:item1 7
user3:item2 9
user3:item3 61、mapper对记录转换成 user1:8 item1
2、排序比较器根据用户id和评分排序，评分是倒排序
3、分组比较器按照用户id分组
4、reducer对同一个用户的推荐物品及评分合并
一、mapper的输入数据是上一步的结果集，转换数据形态。
public class ItemCFStep6Mapper extends Mapper<Text, Text, Text, Text> {
private final Text mkev = new Text();
private final Text mval = new Text();
@Override
protected void map(Text key, Text value, Context context) throws IOException, InterruptedException {
//样本数据：u10004:i1090 253
String[] ss = StringUtils.split(value.toString(), ':');
mkev.set(key.toString() + ":" + ss[1]);
mval.set(ss[0]);
context.write(mkev, mval);
}
}二、排序比较器
public class ItemCFStep6Comparator extends WritableComparator {
public ItemCFStep6Comparator() {
super(Text.class, true);
}
@Override
public int compare(WritableComparable a, WritableComparable b) {
String[] as = StringUtils.split(a.toString(), ":");
String[] bs = StringUtils.split(b.toString(), ":");
int i = as[0].compareTo(bs[0]);
if (i == 0) {
//这里按分值倒序
return Integer.compare(Integer.parseInt(bs[1]), Integer.parseInt(as[1]));
}
return i;
}
}三、分组比较器
public class ItemCFStep6GroupComparator extends WritableComparator {
public ItemCFStep6GroupComparator() {
super(Text.class, true);
}
@Override
public int compare(WritableComparable a, WritableComparable b) {
String[] as = StringUtils.split(a.toString(), ":");
String[] bs = StringUtils.split(b.toString(), ":");
return as[0].compareTo(bs[0]);
}
}四、reducer拼接排好序的所有商品推荐列表
public class ItemCFStep6Reducer extends Reducer<Text, Text, Text, Text> {
private final Text rkey = new Text();
private final Text rval = new Text();
@Override
protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
String[] ss = StringUtils.split(key.toString(), ':');
Iterator<Text> iterator = values.iterator();
StringBuffer sb = new StringBuffer(iterator.next().toString() + ":" + ss[1]);
while (iterator.hasNext()) {
ss = StringUtils.split(key.toString(), ':');
sb.append(",").append(iterator.next().toString() + ":" + ss[1]);
}
rkey.set(ss[0]);
rval.set(sb.toString());
context.write(rkey, rval);
}
}五、最后输出排好序的用户物品推荐列表
user1 item2:9,item1:8,item3:5
user2 item2:8,item3:6,item1:5
user3 item2:9,item1:7,item3:6
六、完整代码详见码云：hadoop-test传送门
/**
* 通过Spark算子实现协同过滤算法的物品推荐
*/
public class ItemUserSpark {
public static void main(String[] args) {
SparkSession ss = SparkSession.builder().master("local")
.appName("item_user")
.getOrCreate();
JavaSparkContext jsc = new JavaSparkContext(ss.sparkContext());
JavaRDD<String> textRdd = jsc.textFile("file:/bigdata/hadoop-test/input/itemCF/test/input/test_data.txt");
/*@step1 加载测试数据转换为二元组rdd：<itemId,<userId,record>>*/
JavaPairRDD<String, Tuple2<String, Integer>> userItemRecordRdd = textRdd.distinct().mapToPair(line -> new Tuple2<>(line.split(",")[0],
new Tuple2<>(line.split(",")[1], UserAction.getRecord(line.split(",")[2]))));
/*@step2 计算各个用户对物品的评分：<itemId,Map<userId,record>>*/
JavaPairRDD<String, HashMap<String, Integer>> userItemRecordCountRdd = userItemRecordRdd.aggregateByKey(new HashMap<>(), new Function2<HashMap<String, Integer>, Tuple2<String, Integer>, HashMap<String, Integer>>() {
private static final long serialVersionUID = 4896422919000738408L;
@Override
public HashMap<String, Integer> call(HashMap<String, Integer> v1, Tuple2<String, Integer> v2) throws Exception {
v1.merge(v2._1, v2._2, Integer::sum);
return v1;
}
}, new Function2<HashMap<String, Integer>, HashMap<String, Integer>, HashMap<String
, Integer>>() {
private static final long serialVersionUID = 2135196741108911099L;
@Override
public HashMap<String, Integer> call(HashMap<String, Integer> v1, HashMap<String, Integer> v2) throws Exception {
for (Map.Entry<String, Integer> entry : v2.entrySet()) {
v1.merge(entry.getKey(), entry.getValue(), Integer::sum);
}
return v1;
}
});
userItemRecordCountRdd.cache();
/*@step3 转换为统计后的用户对物品评分rdd：<itemId,<userId,record>>*/
JavaPairRDD<String, Tuple2<String, Integer>> itemUserRecordRdd = userItemRecordCountRdd.mapPartitionsToPair(new PairFlatMapFunction<Iterator<Tuple2<String, HashMap<String, Integer>>>, String, Tuple2<String, Integer>>() {
private static final long serialVersionUID = -2318624350497226532L;
@Override
public Iterator<Tuple2<String, Tuple2<String, Integer>>> call(Iterator<Tuple2<String, HashMap<String, Integer>>> tuple2Iterator) throws Exception {
List<Tuple2<String, Tuple2<String, Integer>>> ret = new ArrayList<>();
while (tuple2Iterator.hasNext()) {
Tuple2<String, HashMap<String, Integer>> next = tuple2Iterator.next();
for (Map.Entry<String, Integer> entry : next._2.entrySet()) {
ret.add(new Tuple2<>(entry.getKey(), new Tuple2<>(next._1, entry.getValue())));
}
}
return ret.iterator();
}
});
/*@step4 生成物品对物品同现次数的二元组rdd：<<itemId,itemId>, 1>*/
JavaPairRDD<Tuple2<String, String>, Integer> itemItemTmpRdd = userItemRecordCountRdd.mapPartitionsToPair(new PairFlatMapFunction<Iterator<Tuple2<String, HashMap<String, Integer>>>, Tuple2<String, String>, Integer>() {
private static final long serialVersionUID = -2318624350497226532L;
@Override
public Iterator<Tuple2<Tuple2<String, String>, Integer>> call(Iterator<Tuple2<String, HashMap<String, Integer>>> tuple2Iterator) throws Exception {
List<Tuple2<Tuple2<String, String>, Integer>> ret = new ArrayList<>();
while (tuple2Iterator.hasNext()) {
Set<String> itemSet = tuple2Iterator.next()._2.keySet();
Object[] os = itemSet.toArray();
for (int i = 0; i < os.length; i++) {
ret.add(new Tuple2<>(new Tuple2<>(os[i].toString(), os[i].toString()), 1));
for (int j = i + 1; j < os.length; j++) {
ret.add(new Tuple2<>(new Tuple2<>(os[i].toString(), os[j].toString()), 1));
}
}
}
return ret.iterator();
}
});
/*@step5 统计得到物品对物品同现矩阵rdd：<itemId,<itemId, count>>*/
JavaPairRDD<String, Tuple2<String, Integer>> itemItemRdd = itemItemTmpRdd.reduceByKey(Integer::sum)
.mapPartitionsToPair(new PairFlatMapFunction<Iterator<Tuple2<Tuple2<String, String>, Integer>>, String, Tuple2<String, Integer>>() {
private static final long serialVersionUID = 8193629530163757444L;
@Override
public Iterator<Tuple2<String, Tuple2<String, Integer>>> call(Iterator<Tuple2<Tuple2<String, String>, Integer>> tuple2Iterator) throws Exception {
List<Tuple2<String, Tuple2<String, Integer>>> ret = new ArrayList<>();
while (tuple2Iterator.hasNext()) {
Tuple2<Tuple2<String, String>, Integer> mapTuple2 = tuple2Iterator.next();
ret.add(new Tuple2<>(mapTuple2._1._1, new Tuple2<>(mapTuple2._1._2, mapTuple2._2)));
if (!mapTuple2._1._1.equals(mapTuple2._1._2)) {
//如果两个物品不相等，需要再输出一条镜像记录
ret.add(new Tuple2<>(mapTuple2._1._2, new Tuple2<>(mapTuple2._1._1, mapTuple2._2)));
}
}
return ret.iterator();
}
});
itemItemRdd.cache();
System.out.println("**********物品同现矩阵");
itemItemRdd.foreach(t -> System.out.println(t));
/*@step7 用户对物品评分矩阵乘以物品同现矩阵：<itemId, <userId, record>, <itemId, count>>*/
JavaPairRDD<String, Tuple2<Tuple2<String, Integer>, Tuple2<String, Integer>>> join = itemUserRecordRdd.join(itemItemRdd);
/*@step8 得到矩阵行列乘积rdd：<<userId, itemId>, value>*/
JavaPairRDD<Tuple2<String, String>, Integer> userItemMultiRdd = join.mapToPair(new PairFunction<Tuple2<String, Tuple2<Tuple2<String, Integer>, Tuple2<String, Integer>>>, Tuple2<String, String>, Integer>() {
private static final long serialVersionUID = -2951379396437189294L;
@Override
public Tuple2<Tuple2<String, String>, Integer> call(Tuple2<String, Tuple2<Tuple2<String, Integer>, Tuple2<String, Integer>>> t) throws Exception {
return new Tuple2<>(new Tuple2<>(t._2._1._1, t._2._2._1), t._2._1._2 * t._2._2._2);
}
});
/*@step9 求和得到用户对每个物品的物品的推荐评分矩阵：<<userId, itemId>, value>*/
JavaPairRDD<Tuple2<String, String>, Integer> userItemMultiRecordRdd = userItemMultiRdd.reduceByKey(Integer::sum);
System.out.println("**********用户对物品推荐矩阵");
userItemMultiRecordRdd.foreach(t -> System.out.println(t));
/*@step10 转换成二元组rdd：<<userId, value>, itemId>*/
JavaPairRDD<Tuple2<String, Integer>, String> userRecordItemRdd = userItemMultiRecordRdd.mapToPair(new PairFunction<Tuple2<Tuple2<String, String>, Integer>, Tuple2<String, Integer>, String>() {
private static final long serialVersionUID = 3046037895848145693L;
@Override
public Tuple2<Tuple2<String, Integer>, String> call(Tuple2<Tuple2<String, String>, Integer> t) throws Exception {
return new Tuple2<>(new Tuple2<>(t._1._1, t._2), t._1._2);
}
});
//排序聚合后的推荐列表rdd：<userId, string>
JavaPairRDD<String, String> userItemsRdd = userRecordItemRdd.sortByKey(new UserRecordCompare())
/*@step11 自定义比较排序器，先按照userId自然排序，然后按照评分倒序排序*/
.mapToPair(new PairFunction<Tuple2<Tuple2<String, Integer>, String>, String, Tuple2<String, Integer>>() {
private static final long serialVersionUID = 2210128857753846055L;
@Override
public Tuple2<String, Tuple2<String, Integer>> call(Tuple2<Tuple2<String, Integer>, String> t) throws Exception {
return new Tuple2<>(t._1._1, new Tuple2<>(t._2, t._1._2));
}
/*@step12 聚合排序后的每个用户的推荐列表*/
}).aggregateByKey("", new Function2<String, Tuple2<String, Integer>, String>() {
private static final long serialVersionUID = -2801745541613060736L;
@Override
public String call(String v1, Tuple2<String, Integer> v2) throws Exception {
return v1 + "," + v2._1 + ":" + v2._2;
}
}, new Function2<String, String, String>() {
private static final long serialVersionUID = -81211973327382009L;
@Override
public String call(String v1, String v2) throws Exception {
return v1 + v2;
}
});
System.out.println("*****用户物品推荐列表");
userItemsRdd.foreach(t -> System.out.println(t));
}
public static class UserRecordCompare implements Serializable, Comparator<Tuple2<String, Integer>> {
private static final long serialVersionUID = 1254251593402832337L;
@Override
public int compare(Tuple2<String, Integer> o1, Tuple2<String, Integer> o2) {
int i = o1._1.compareTo(o2._1);
if (i == 0) {
i = o2._2.compareTo(o1._2);
}
return i;
}
}
}