利用Hadoop实现求共同好友的示例详解

public class FirstMapper extends Mapper<LongWritable,Text,Text,Text> {

    @Override
    protected void map(LongWritable key, Text value, Context context) throws ioException, InterruptedException {
        String val = value.toString();
        String[] split = val.split(":");
        //A:B,C,D,F,E,O  拆分后，左边是原用户，右边是好友
        String user = split[0];
        String friends = split[1];
        String[] friendLists = friends.split(",");
        //Map1 输出的结果为 ：
        
        for(String str :friendLists ){
            context.write(new Text(str),new Text(user));
        }
    }

}

public class FirstReducer extends Reducer<Text,Text,Text,Text> {

    @Override
    protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
        StringBuffer stringBuffer = new StringBuffer();
        for (Text text : values){
            stringBuffer.append(text).append("-");
        }
        //最终写出去的数据格式为： A-E B ......
        context.write(new Text(stringBuffer.toString()),key);
    }

}

public class FirstJob {

    public static void main(String[] args) throws Exception {

        //1、获取job
        Configuration configuration = new Configuration();
        Job job = Job.getInstance(configuration);

        //2、设置jar路径
        job.setJarByClass(FirstJob.class);

        //3、关联mapper 和 Reducer
        job.setMapperClass(FirstMapper.class);
        job.setReducerClass(FirstReducer.class);

        //4、设置 map输出的 key/val 的类型
        job.setMapOutpuTKEyClass(Text.class);
        job.setMapOutputValueClass(Text.class);

        //5、设置最终输出的key / val 类型
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);

        //6、设置最终的输出路径
        String inputPath = "F:\\网盘\\csv\\friends.txt";
        String outPath = "F:\\网盘\\csv\\friends1";

        FileInputFORMat.setInputPaths(job,new Path(inputPath));
        FileOutputFormat.setOutputPath(job,new Path(outPath));

        // 7 提交job
        boolean result = job.waitForCompletion(true);
        System.exit(result ? 0 : 1);
    }

}

public class SecondMapper extends Mapper<LongWritable,Text,Text,Text> {

    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {

        // I-K-C-B-G-F-H-O-D-    A  阶段1的文件输出格式
        
        //需要将左边的数据进行两两拆分，与V进行组合输出
        String val = value.toString();
        String[] split = val.split("\t");

        String v2 = split[1];
        String[] allUsers = split[0].split("-");
        Arrays.sort(allUsers);

        for(int i=0;i<allUsers.length-1;i++){
            for(int j=i+1;j<allUsers.length;j++){
                context.write(new Text(allUsers[i] + "-" + allUsers[j]),new Text(v2));
            }
        }
    }
}

public class SecondReducer extends Reducer<Text,Text,Text,Text> {

    @Override
    protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
        //上一步输出的结果：
        
        //只需要将相同的key的Val进行组合即可,即 ： A-B C-D，A-E C-D
        StringBuffer stringBuffer = new StringBuffer();
        for (Text text :values ){
            stringBuffer.append(text.toString()).append("-");
        }
        context.write(key,new Text(stringBuffer.toString()));
    }

}

public class SecondJob {
    public static void main(String[] args) throws Exception {

        //1、获取job
        Configuration configuration = new Configuration();
        Job job = Job.getInstance(configuration);

        //2、设置jar路径
        job.setJarByClass(SecondJob.class);

        //3、关联mapper 和 Reducer
        job.setMapperClass(SecondMapper.class);
        job.setReducerClass(SecondReducer.class);

        //4、设置 map输出的 key/val 的类型
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(Text.class);

        //5、设置最终输出的key / val 类型
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);

        //6、设置最终的输出路径
        String inputPath = "F:\\网盘\\csv\\friends1\\part-r-00000";
        String outPath = "F:\\网盘\\csv\\friends2";

        FileInputFormat.setInputPaths(job,new Path(inputPath));
        FileOutputFormat.setOutputPath(job,new Path(outPath));

        // 7 提交job
        boolean result = job.waitForCompletion(true);
        System.exit(result ? 0 : 1);
    }

}