MapReduce高级案例⑤

共同好友计算

以下是好友列表数据，冒号前是一个用户，冒号后是该用户的所有好友（数据中的好友关系是单向的）
[collapse title="数据"]

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

[/collapse]

第一次处理

[collapse title="输出"]

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

[/collapse]
[collapse title="ruaDriver01"]

package com.kami.demo05;

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.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

/**
 * @version v 1.0
 * @Author kamisamak
 * @Date 2020/6/16
 */
public class ruaDriver01 {
    public static void main(String[] args) throws Exception {
        Configuration configuration = new Configuration();
        Job job = Job.getInstance(configuration);
        job.setJarByClass(ruaDriver01.class);
        job.setMapperClass(ruaMapper01.class);
        job.setReducerClass(ruaReducer01.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(Text.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);
        FileInputFormat.setInputPaths(job, new Path("data\\d05\\rua.txt"));
        FileOutputFormat.setOutputPath(job, new Path("output\\d11"));
        boolean result = job.waitForCompletion(true);
        System.exit(result ? 0 : 1);
    }
}

[/collapse]
[collapse title="ruaMapper01"]

package com.kami.demo05;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;

/**
 * @version v 1.0
 * @Author kamisamak
 * @Date 2020/6/16
 */
    public class ruaMapper01 extends Mapper {
    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        String line = value.toString();
        String[] fileds = line.split(":");
        //获取person和好友
        String person = fileds[0];
        String[] friends = fileds[1].split(",");
        for (String friend : friends) {
            // 输出 <好友，人>
            context.write(new Text(friend), new Text(person));
        }
    }
}

[/collapse]
[collapse title="ruaReducer01"]

package com.kami.demo05;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;

/**
 * @version v 1.0
 * @Author kamisamak
 * @Date 2020/6/16
 */
public class ruaReducer01 extends Reducer {
    @Override
    protected void reduce(Text key, Iterable values, Context context) throws IOException, InterruptedException {
        StringBuffer sb = new StringBuffer();
        for (Text value : values) {
            sb.append(value).append(",");
        }
        context.write(key,new Text(sb.toString()));
    }
}

[/collapse]

第二次处理

[collapse title="输出"]


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

[/collapse]
[collapse title="ruaDriver02"]

package com.kami.demo05;

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.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

/**
 * @version v 1.0
 * @Author kamisamak
 * @Date 2020/6/16
 */
public class ruaDriver02 {
    public static void main(String[] args) throws Exception {
        Configuration configuration = new Configuration();
        Job job = Job.getInstance(configuration);
        job.setJarByClass(ruaDriver02.class);
        job.setMapperClass(ruaMapper02.class);
        job.setReducerClass(ruaReducer02.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(Text.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);
        FileInputFormat.setInputPaths(job, new Path("output\\d11"));
        FileOutputFormat.setOutputPath(job, new Path("output\\d12"));
        boolean result = job.waitForCompletion(true);
        System.exit(result ? 0 : 1);
    }
}

[/collapse]
[collapse title="ruaMapper02"]

package com.kami.demo05;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;
import java.util.Arrays;

/**
 * @version v 1.0
 * @Author kamisamak
 * @Date 2020/6/16
 */
public class ruaMapper02  extends Mapper {
    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        String line = value.toString();
        String[] fps = line.split("\t");

        String friend = fps[0];
        String[] persons = fps[1].split(",");
        Arrays.sort(persons);

        for (int i = 0; i < persons.length - 1; i++) {
            for (int j = i + 1; j < persons.length; j++) {
                //人-人，好友
                context.write(new Text(persons[i] + "-" + persons[j]), new Text(friend));
            }
        }
    }
}

[/collapse]
[collapse title="ruaReducer02"]

package com.kami.demo05;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;

/**
 * @version v 1.0
 * @Author kamisamak
 * @Date 2020/6/16
 */
public class ruaReducer02 extends Reducer {
    @Override
    protected void reduce(Text key, Iterable values, Context context) throws IOException, InterruptedException {
        StringBuffer sb = new StringBuffer();
        for (Text friend : values) {
            sb.append(friend).append(" ");
        }
        context.write(key, new Text(sb.toString()));
    }
}

[/collapse]

题目来源:https://www.cnblogs.com/frankdeng/p/9255931.html

MapReduce高级案例⑤

共同好友计算

文章评论