数据挖掘--协同过滤算法，基于集合交集相似性计算的mapreduce算法设计

第一个过程程的mapreduce函数：

import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.StringTokenizer;

public class MapReduce extends MapperReduce{

public MapReduce(){
super();
}

public void printf(List<String> list1,List<String> list2) throws IOException{
if((!list1.isEmpty()) && (!list2.isEmpty())){
Iterator<String> Iter1=list1.iterator();
while(Iter1.hasNext()){
String tmp1=Iter1.next();
Iterator<String> Iter2=list2.iterator();
while(Iter2.hasNext()){
String tmp2=Iter2.next();
WriteReducer(tmp1,tmp2);
}
}
}
}

//map函数开始
public void Mapper(String key,String value) throws IOException{
//key为行偏移量，value为每一行的值
String[] count=value.split(" ");
if(!count[1].trim().equals("")){
WriteMaper(count[1].trim(),count[0].trim()+" "+"1");
WriteMaper(count[1].trim(),count[0].trim()+" "+"2");
}
}
//map函数结束

//reduce函数开始
public void Reducer(String key,Iterator<String> value) throws IOException{
List<String> list1=new ArrayList<String>();
List<String> list2=new ArrayList<String>();
while(value.hasNext()){
String tmp=value.next();
String[] tmp1=tmp.split(" ");
if(tmp1[1].equals("1")){
list1.add(tmp1[0]);
}else{
list2.add(tmp1[0]);
}
}
printf(list1,list2);
}
//reduce函数结束
public static void main(String[] args) throws IOException {
MapReduce a=new MapReduce();
a.IterMapper();
a.IterReducer();
}
}

本程序实例是：单表自连接，连接字段是 user

输入：
star1 user1

star2 user1

star3 user1

star3 user2

输出：

star1 star1

star1 star2

star1 star3

star2 star1

star2 star2

star2 star3

star3 star1

star3 star2

star3 star3

star3 star3

第二个过程程的mapreduce函数：

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;

public class MapReduce extends MapperReduce{

public MapReduce(){
super();
}

//map函数开始
public void Mapper(String key,String value) throws IOException{
String[] content=value.split(" ");
WriteMaper(content[0],content[1]);
}
//map函数结束

//reduce函数开始
public void Reducer(String key,Iterator<String> value) throws IOException{
List<String> list=new ArrayList<String>();
while(value.hasNext()){
list.add(value.next());
}
Set<String> set = new HashSet<String>(list);
Map<String,Integer> map=new HashMap<String,Integer>();
Iterator<String> IterSet=set.iterator();
while(IterSet.hasNext()){
map.put(IterSet.next(),0);
}
for(int i=0;i<list.size();i++){
map.put(list.get(i),map.get(list.get(i))+1);
}
Iterator<String> It = map.keySet().iterator();
while(It.hasNext())
{
String tmp=It.next();
if(!tmp.equals(key)){
WriteReducer(key,tmp+" "+String.valueOf(((float)map.get(tmp))/map.get(key)));
}
}
}
//reduce函数结束
public static void main(String[] args) throws IOException {
MapReduce a=new MapReduce();
a.IterMapper();
a.IterReducer();
}
}

输入：

star1 star1

star1 star2

star1 star3

star2 star1

star2 star2

star2 star3

star3 star1

star3 star2

star3 star3

star3 star3

输出：

star1 star2 1.0

star1 star3 1.0

star2 star1 1.0

star2 star3 1.0

star3 star1 0.5

star3 star2 0.5