Mapreduce案例之找共同好友

数据准备：

A:B,C,D,F,E,O
B:A,C,E,K
C:F,A,D,I
D:A,E,F,L
E:B,C,D,M,L
F:A,B,C,D,E,O,M
G:A,C,D,E,F
H:A,C,D,E,O
I:A,O
J:B,O
K:A,C,D
L:D,E,F
M:E,F,G
O:A,H,I,J

需求：

1.先求出A、B、C、….等是谁的好友

2.求出哪些人两两之间有共同好友，及他俩的共同好友都有谁？

需求解读：

1.有题目可得，该关系为单项关系可以理解为关注，即A关注的为BCDEF，B关注的为AK，所以求A B C...是谁的关注，即需要将冒号后边的作为key，前边的作为value进行map，在reduce的时候在合并即可。

2.求两两之间的共同关注，这时我们需要借助1题产生的结果文件，map是从后边的value值两两组合生成两两关系。作为新的key值输入，新的value为1题的key。

reduce的时候，将结果合并即可

源代码如下

第一题：

package Demo5;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
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.Reducer.Context;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
/**
 * @author 星际毁灭
 * 找出ABC等人的好友
 * */
import Demo4.Log1;

public class Friend1 {
    public static class Map extends Mapper<Object , Text , Text , Text>{  
        private static Text newKey=new Text();  
        private static final IntWritable one = new IntWritable(1);
        public void map(Object key,Text value,Mapper<Object, Text, Text, Text>.Context context) throws IOException, InterruptedException{  
            String line=value.toString(); 
            String[] array=line.split(":");  //数据实例  A:B,C,D,F,E,O
            String first=array[0];   //冒号前边部分
            String second=array[1];  //冒号后边部分
            String[] friend=second.split(",");  //冒号后边部分在切割
            System.out.println(first+"	"+second);
            for(int i=0;i<friend.length;i++) {    
                context.write(new Text(friend[i]),new Text(first));  //好友关系，即源文件的反向
            }
            
         }
    }  
    public static class Reduce extends Reducer<Text, Text, Text, Text>{  
        public void reduce(Text key,Iterable<Text> values,Context context) throws IOException, InterruptedException{  
            String res="";  
            for(Text val:values) {
                res+=val.toString()+",";  //将结果合并
            }
            context.write(key,new Text(res));
        } 
     }  
     public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException{  
             System.setProperty("hadoop.home.dir", "H:\文件\hadoop\hadoop-2.6.4");
            Configuration conf=new Configuration();  
            Path in=new Path("hdfs://192.168.6.132:9000/wys/in/friend1.txt");  
            Path out=new Path("hdfs://192.168.6.132:9000/wys/out/friend1");  
            
            Job job =new Job(conf,"OneSort");  
            
            FileInputFormat.addInputPath(job,in);  
            FileOutputFormat.setOutputPath(job,out);  
            job.setJarByClass(Friend1.class);  
            job.setMapperClass(Friend1.Map.class);
            job.setReducerClass(Friend1.Reduce.class);
            
            job.setOutputKeyClass(Text.class);
            job.setOutputValueClass(Text.class);
            job.waitForCompletion(true);
            
            
            System.exit(job.waitForCompletion(true) ? 0 : 1);       
      } 

}

第二题：

package Demo5;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
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.Reducer.Context;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

/**
 * @author 王翌淞
 * 
 * 求出他们的共同好友
 * */
public class Friend2 {
    public static class Map extends Mapper<Object , Text , Text , Text>{  
        private static Text newKey=new Text();  
        private static final IntWritable one = new IntWritable(1);
        public void map(Object key,Text value,Mapper<Object, Text, Text, Text>.Context context) throws IOException, InterruptedException{  
            String line=value.toString(); 
            String[] array=line.split("	");
            String first=array[0];
            String second=array[1];
            String[] friend=second.split(",");
            System.out.println(first+"	"+second);
            
            //通过两层的for循环，A的好友进行组合，有顺序
            for(int i=0;i<friend.length;i++) {   
                for(int j=i+1;j<friend.length;j++) {
                    context.write(new Text(friend[i]+"-"+friend[j]),new Text(first));  //正序关系
                    context.write(new Text(friend[j]+"-"+friend[i]),new Text(first));  //倒序关系
                }
                
            }
            //context.write(new Text(friend[i]),new Text(first));
            
         }
    }  
    public static class Reduce extends Reducer<Text, Text, Text, Text>{  
        public void reduce(Text key,Iterable<Text> values,Context context) throws IOException, InterruptedException{  
            String res="";        
            for(Text val:values) {
                res+=val.toString()+" ";
            }
            context.write(key,new Text(res));
           } 
     }  
     public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException{  
             System.setProperty("hadoop.home.dir", "H:\文件\hadoop\hadoop-2.6.4");
            Configuration conf=new Configuration();  
            Path in=new Path("hdfs://192.168.6.132:9000/wys/out/friend1/part-r-00000");  
            Path out=new Path("hdfs://192.168.6.132:9000/wys/out/friend2");  
            
            Job job =new Job(conf,"OneSort");  
            
            FileInputFormat.addInputPath(job,in);  
            FileOutputFormat.setOutputPath(job,out);  
            job.setJarByClass(Friend2.class);  
            job.setMapperClass(Friend2.Map.class);
            job.setReducerClass(Friend2.Reduce.class);
            
            job.setOutputKeyClass(Text.class);
            job.setOutputValueClass(Text.class);
            job.waitForCompletion(true);
            
            
            System.exit(job.waitForCompletion(true) ? 0 : 1);       
      } 

}