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

 

//Class1

package com;

import org.apache.hadoop.conf.Configuration;

import org.apache.hadoop.fs.Path;

import org.apache.hadoop.io.LongWritable;

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.net.URISyntaxException;

/**

 *

 * 2017-08-17

 *

 * Class1

 *

 */

public class Class1 {

    public static class mapper extends Mapper<LongWritable,Text,Text,Text> {

        

        Text key_ = new Text();

        Text value_ = new Text();

        

        @Override

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

           

            String line = value.toString();

            String[] words = line.split(":");

            String people = words[0];

            String[] friends = words[1].split(",");

            

            for(String friend : friends){

                

                key_.set(friend);

                value_.set(people);

                

                //(朋友,人)OR(朋友,谁的朋友)

                context.write(key_,value_);

                

            }

        }

    }

    public static class reducer extends Reducer<Text,Text,Text,Text>{

        

        @Override

        protected void reduce(Text friend, Iterable<Text> iterable, Context context) throws IOException, InterruptedException {

               

            StringBuffer people = new StringBuffer();

            

            for (Text text : iterable) {

                

                　　if(people.length()!=0){

                    

                   　　　　 people.append(",");

                    

               　　 }

                

                　　people.append(text);

            }

            

            　　//Key(朋友),Value(谁的朋友)

            　　context.write(friend,new Text(people.toString()));

        }

        

    }

    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException, URISyntaxException {

        

        Configuration config = new Configuration();

        config.set("fs.defaultFS", "hdfs://192.168.0.117:9000");

        config.set("yarn.resourcemanager.hostname", "192.168.0.117");

        

        Job job = Job.getInstance(config);

       

        job.setMapperClass(mapper.class);

        job.setReducerClass(reducer.class);

       

        job.setMapOutputKeyClass(Text.class);

        job.setMapOutputValueClass(Text.class);

       

        job.setOutputKeyClass(Text.class);

        job.setOutputValueClass(Text.class);

        

        

        FileInputFormat.setInputPaths(job, new Path("/friend.txt"));

        FileOutputFormat.setOutputPath(job, new Path("/Out"));

        

        boolean b = job.waitForCompletion(true);

      

　　 if(b){

            　　System.out.println("Success");

        }else{

            　　System.out.println("Error");

        }

    }

}

 第一次输出:

A    I,K,C,B,G,F,H,O,D

B    A,F,J,E

C    A,E,B,H,F,G,K

D    G,C,K,A,L,F,E,H

E    G,M,L,H,A,F,B,D

F    L,M,D,C,G,A

G    M

H    O

I    O,C

J    O

K    B

L    D,E

M    E,F

O    A,H,I,J,F

 

 

 

 

 

 

 

 

 

 

//Class2

 

 

 

package com;

import org.apache.hadoop.conf.Configuration;

import org.apache.hadoop.fs.Path;

import org.apache.hadoop.io.LongWritable;

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.net.URISyntaxException;

import java.util.Arrays;

/**

 *

 * 2017-8-17

 *

 * Class2

 *

 */

 

public class Class2 {

    public static class mapper extends Mapper<LongWritable,Text,Text,Text> {

       

        Text key_ = new Text();

        Text value_ = new Text();

        

        @Override

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

            

            //Key-Value分割

            String line = value.toString();

            String[] words = line.split("\t");

            

            //朋友

            String friend = words[0];

            //谁的朋友,那个谁!

            String[] people = words[1].split(",");

            

            //排序

            Arrays.sort(people);

            

            for(int i = 0 ; i<people.length-2;i++){

                

                for(int j=i+1;j<people.length-1;j++){

                    

                    //谁跟谁的朋友          people-people 的共同好友是: friend

                    context.write(new Text(people[i]+"-"+people[j]),new Text(friend));

                    

                }

                

            }

            

        }

    }

    public static class reducer extends Reducer<Text,Text,Text,Text>{

        

        @Override

        protected void reduce(Text people, Iterable<Text> iterable, Context context) throws IOException, InterruptedException {

           

            StringBuffer friend = new StringBuffer();

           

            for (Text text : iterable) {

                

                //以空格分割,追加共同好友

                friend.append(text).append(" ");

                

            }

            

            context.write(new Text(people+"的共同好友是:"),new Text(friend.toString()));

            

        }

        

    }

    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException, URISyntaxException {

        

        Configuration config = new Configuration();

        config.set("fs.defaultFS", "hdfs://192.168.0.117:9000");

        config.set("yarn.resourcemanager.hostname", "192.168.0.117");

        

        Job job = Job.getInstance(config);

        

        job.setMapperClass(mapper.class);

        job.setReducerClass(reducer.class);

        

        job.setMapOutputKeyClass(Text.class);

        job.setMapOutputValueClass(Text.class);

        

        job.setOutputKeyClass(Text.class);

        job.setOutputValueClass(Text.class);

        

        FileInputFormat.setInputPaths(job, new Path("/Out"));

        FileOutputFormat.setOutputPath(job, new Path("/Out1"));

        

        boolean b = job.waitForCompletion(true);

        if(b){

            System.out.println("Success");

        }else{

            System.out.println("Error");

        }

}}

 

最终结果:

A-B的共同好友是:    C E

A-C的共同好友是:    F D

A-D的共同好友是:    E F

A-E的共同好友是:    B C D

A-F的共同好友是:    C D B E O

A-G的共同好友是:    D E F C

A-H的共同好友是:    E O C D

A-I的共同好友是:    O

A-K的共同好友是:    D

A-L的共同好友是:    F E

B-C的共同好友是:    A

B-D的共同好友是:    E A

B-E的共同好友是:    C

B-F的共同好友是:    E A C

B-G的共同好友是:    C E A

B-H的共同好友是:    E C A

B-I的共同好友是:    A

B-K的共同好友是:    A

B-L的共同好友是:    E

C-D的共同好友是:    F A

C-E的共同好友是:    D

C-F的共同好友是:    D A

C-G的共同好友是:    F A D

C-H的共同好友是:    A D

C-I的共同好友是:    A

C-K的共同好友是:    D A

C-L的共同好友是:    F

D-F的共同好友是:    E A

D-G的共同好友是:    A E F

D-H的共同好友是:    A E

D-I的共同好友是:    A

D-K的共同好友是:    A

D-L的共同好友是:    F E

E-F的共同好友是:    C D B

E-G的共同好友是:    D C

E-H的共同好友是:    D C

E-K的共同好友是:    D

F-G的共同好友是:    C E D A

F-H的共同好友是:    C A D E O

F-I的共同好友是:    A O

F-K的共同好友是:    D A

F-L的共同好友是:    E

G-H的共同好友是:    D E C A

G-I的共同好友是:    A

G-K的共同好友是:    A D

G-L的共同好友是:    F E

H-I的共同好友是:    A O

H-K的共同好友是:    A D

H-L的共同好友是:    E

I-K的共同好友是:    A