0
我想運行下面的代碼，找到兩個人之間的共同朋友。輸入如下hadoop mapreduce common friends reducer spillage
A : B C D
B : A C D E
C : A B D E
D : A B C E
E : B C D
我不能在輸出文件中得到任何輸出，也沒有例外。
請在下面找到我的代碼，
public class Friend {
public static class Mapperfriend extends Mapper<Object, Text, Text, Text>{
private Text vendor = new Text();
@Override
protected void map(Object key, Text value, Context context) throws IOException, InterruptedException {
StringTokenizer tokenizer = new StringTokenizer(value.toString(), "\n");
String line = null;
String[] lineArray = null;
String[] friendArray = null;
String[] tempArray = null;
while(tokenizer.hasMoreTokens()){
line = tokenizer.nextToken();
lineArray = line.split(":");
friendArray = lineArray[1].split(" ");
tempArray = new String[2];
for(int i = 0; i < friendArray.length; i++){
tempArray[0] = friendArray[i];
tempArray[1] = lineArray[0];
Arrays.sort(tempArray);
context.write(new Text(tempArray[0] + " " + tempArray[1]), new Text(lineArray[1]));
}
}
}}
public static class ReducerFriend extends Reducer<Text,Text,Text,Text>{
@Override
protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
Text[] texts = new Text[2];
int index = 0;
for(Text val: values)
{
texts[index++] = new Text(val);
}
String[] list1 = texts[0].toString().split(" ");
String[] list2 = texts[1].toString().split(" ");
List<String> list = new LinkedList<String>();
for(String friend1 : list1){
for(String friend2 : list2){
if(friend1.equals(friend2)){
list.add(friend1);
}
}
}
StringBuffer sb = new StringBuffer();
for(int i = 0; i < list.size(); i++){
sb.append(list.get(i));
if(i != list.size() - 1)
sb.append(" ");
}
context.write(key, new Text(sb.toString()));
}
}
/**
* @param args the command line arguments
*/
public static void main(String[] args) throws Exception {
// TODO code application logic here
Configuration conf = new Configuration();
Job job = Job.getInstance(conf, "Friends");
job.setJarByClass(Friend.class);
job.setMapperClass(Mapperfriend.class);
job.setReducerClass(ReducerFriend.class);
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(Text.class);
FileInputFormat.addInputPath(job, new Path(args[0]));
FileOutputFormat.setOutputPath(job, new Path(args[1]));
System.exit(job.waitForCompletion(true) ? 0 : 1);
}
}