使用MapReduce实现二度人脉搜索算法详解大数据

a    b 
b    c 
a    c 
b    d 
c    e 
e    c 
e    f

import org.apache.hadoop.conf.Configuration; 
import org.apache.hadoop.fs.Path; 
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.FileInputFormat; 
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; 
import java.io.IOException; 
import java.util.HashSet; 
import java.util.Random; 
import java.util.Set; 
public class De2Friends { 
public static class De2Mapper1 extends Mapper<Object,Text,Text,Text>{ 
@Override 
protected void map(Object key, Text value, Context context) throws 
IOException, InterruptedException { 
String line =value.toString(); 
String[] strArr = line.split("/t"); 
if(strArr.length==2) { 
//关注的人 
context.write(new Text(strArr[0]), new Text("1" + strArr[1])); 
//被关注的人 
context.write(new Text(strArr[1]), new Text("0" + strArr[0])); 
} 
} 
} 
public static class De2Reducer1 extends Reducer<Text,Text,Text,Text> { 
@Override 
protected void reduce(Text key, Iterable<Text> values, Context context) 
throws IOException, InterruptedException { 
Set<String> follows= new HashSet<String>(); 
Set<String> fans=new HashSet<String>(); 
for(Text val :values ){ 
String friend =val.toString(); 
if(friend.startsWith("1")){ 
context.write(key,new Text(friend));//输出用户已经关注的人,一度人脉 
follows.add(friend.substring(1)); 
} 
if(friend.startsWith("0")){ 
fans.add(friend.substring(1)); 
} 
} 
for(String fan : fans) 
for(String follow:follows) { 
if (!fan.equals(follow)) { 
context.write(new Text(fan),new Text("2"+follow)); 
} 
} 
} 
} 
public static class De2Mapper2 extends  Mapper<Object,Text,Text,Text>{ 
@Override 
protected void map(Object key, Text value, Context context) throws IOException, InterruptedException { 
String line =value.toString(); 
String[] strArr=line.split("/t"); 
if(strArr.length==2) { 
context.write(new Text(strArr[0]), new Text(strArr[1]));//输出用户的一度好友和二度好友 
            } 
} 
} 
public static class De2Reducer2 extends Reducer<Text,Text,Text,Text>{ 
@Override 
protected void reduce(Text key, Iterable<Text> values, Context context) 
throws IOException, InterruptedException { 
Set<String> firstFriend = new HashSet<String>(); 
Set<String> secondFriend =new HashSet<String>(); 
for(Text val:values){ 
String friend =val.toString(); 
if(friend.contains("1")){ 
firstFriend.add(friend.substring(1)); 
} 
if(friend.contains("2")){ 
secondFriend.add(friend.substring(1)); 
} 
} 
for(String second:secondFriend) { 
if(!(firstFriend.contains(second))) 
context.write(key,new Text(second)); //输出好友的二度人脉 
                } 
} 
} 
public static void main(String[] args) throws Exception{ 
System.setProperty("hadoop.home.dir","E://softs//majorSoft//hadoop-2.7.5"); 
Configuration conf =new Configuration(); 
conf.set("mapreduce.app-submission.cross-platform", "true"); 
Path fileInput = new Path("hdfs://mycluster/testFile/qq.txt"); 
Path tempDir = new Path("hdfs://mycluster/output/deg2friend-temp-" + Integer.toString(new Random().nextInt(Integer.MAX_VALUE))); 
Path fileOutput = new Path("hdfs://mycluster/output/qq"); 
Job job = Job.getInstance(conf,"de2Firend"); 
job.setJar("E://bigData//hadoopDemo//out//artifacts//wordCount_jar//hadoopDemo.jar"); 
job.setJarByClass(De2Friends.class); 
job.setMapperClass(De2Mapper1.class); 
job.setReducerClass(De2Reducer1.class); 
job.setMapOutputKeyClass(Text.class); 
job.setMapOutputValueClass(Text.class); 
job.setNumReduceTasks(1); 
job.setOutputKeyClass(Text.class); 
job.setOutputValueClass(Text.class); 
FileInputFormat.addInputPath(job,fileInput); 
FileOutputFormat.setOutputPath(job,tempDir); 
job.waitForCompletion(true);//必须有，感觉是等job执行完才让job2执行的效果，即阻塞吧 
Job job2 = Job.getInstance(conf,"de2Firend"); 
job2.setJar("E://bigData//hadoopDemo//out//artifacts//wordCount_jar//hadoopDemo.jar"); 
job2.setJarByClass(De2Friends.class); 
job2.setMapperClass(De2Mapper2.class); 
job2.setReducerClass(De2Reducer2.class); 
job2.setMapOutputKeyClass(Text.class); 
job2.setMapOutputValueClass(Text.class); 
job2.setOutputKeyClass(Text.class); 
job2.setOutputValueClass(Text.class); 
FileInputFormat.addInputPath(job2,tempDir); 
FileOutputFormat.setOutputPath(job2,fileOutput); 
System.exit(job2.waitForCompletion(true)?0:1); 
} 
}

a    d 
b    e 
b    f 
c    f