大数据技术基础教程 课件 第11章 好友案例分析

第11章好友推荐案例分析《大数据技术基础教程》学习目标/Target

了解好友推荐项目需求和目标

掌握项目所需的预备知识实践如何编写MapReduce程序，完成QQ好友推荐操作章节概述/

Summary近十几年间，随着在线社交网络的蓬勃发展，研究人员才开始有机会在大量现实数据的基础上对社交影响力进行建模和分析，并取得了丰硕的研究成果和广泛的应用价值。在这上千万用户的社交群体里，寻找潜在好友进行推荐，成为社交网络分析的关键问题之一。社交网络中的好友是指在社交网络中出现在用户联系人列表中的用户，它是一种广义上的朋友，既可以是Facebook中的好友，也可是新浪微博中的关注用户等。为社交网络用户推荐好友就是帮助用户在社交网络中找到她们感兴趣的用户，进而添加到自己的联系人列表之中，方便进一步的交流及商业应用。本章主要介绍了如何编写MapReduce程序，完成QQ好友推荐操作的相关知识目录/Contents01020304任务需求实验前HDFS平台的准备工作IDEA下建立Maven项目projectFriendRecommend程序实现05运行程序与结果验证任务需求11.111.1任务需求通过mapreduce计算，为每个QQ用户推荐好友。数据集第一列为QQ用户，其他列为该用户的QQ好友。其中xiaoming、laowang、xiaohua、lingling、xiaogang、meimei和xiaolan是选中的7位QQ用户，每位用户后面的人员，是该用户的QQ好友。xiaoming laowang xiaohua linglinglaowang xiaoming xiaolanxiaohua xiaoming xiaogang xiaolanlingling xiaoming xiaogang xiaolan meimeixiaogang xiaohua xiaolan linglingmeimei xiaolan linglingxiaolan xiaohua laowang lingling meimei实验前HDFS平台的准备工作11.211.2实验前HDFS平台的准备工作11.2.1启动Hadoop平台启动Hadoop平台11.2实验前HDFS平台的准备工作11.2.2

实验前数据路径的准备11.2实验前HDFS平台的准备工作11.2.3实验程序每次执行前，运行结果输出路径的准备每次实验程序运行结果会存储在HDFS平台的/root/experiment/output/路径下，所以在程序执行前，该路径不能存在，如果存在，需要进行删除。IDEA下建立Maven项目project11.311.3IDEA下建立Maven项目project1)建立新项目1.打开IDEA工具。2.确定要建立的项目类型11.3IDEA下建立Maven项目project1)建立新项目3.填入工程信息4.确认项目信息，完成项目创建11.3IDEA下建立Maven项目project1)建立新项目5.此时进入IDEA的开发界面。如果在开发界面的上方弹出“TipoftheDay”窗口，点出”Close”按钮，关闭该窗口即可。6.在右下角弹出的对话框中，选择EnableAuto-Import（如未弹出该对话框请忽略此步骤）。7.显示IDEA开发环境的主窗口。11.3IDEA下建立Maven项目project2)配置pom.xml文件11.3IDEA下建立Maven项目project3)查看Hadoop工程的Maven依赖包FriendRecommend程序实现11.411.4FriendRecommend程序实现11.4.1数据准备11.4FriendRecommend程序实现11.4.2编写单表关联的MapReduce程序11.4FriendRecommend程序实现11.4.3建立FirstJob类文件，编写FirstJob类package

experiment;

import

org.apache.hadoop.io.IntWritable;

import

org.apache.hadoop.io.Text;

import

org.apache.hadoop.mapreduce.Mapper;

import

org.apache.hadoop.mapreduce.Reducer;

import

org.apache.hadoop.util.StringUtils;

import

java.io.IOException;

public

class

FirstJob

{

public

static

class

Fof

extends

Text{

public

Fof(){

super();

}

public

Fof(String

a，String

b){

super(getFof(a，

b));

}

/*字符串比较小的放前面*/

public

static

String

getFof(String

a，String

b){

int

r

=pareTo(b);

if(r<0){

return

a+"\t"+b;

}else{

return

b+"\t"+a;

}

}

}

static

class

FofMapper

extends

Mapper<Text，

Text，

Fof，

IntWritable>

{

protected

void

map(Text

key，

Text

value，Context

context)

throws

IOException，

InterruptedException

{

/*获取key的值*/

String

user

=key.toString();

/*字符串分割*/

String[]

friends

=StringUtils.split(value.toString()，

'\t');

/*遍历数据*/

11.4FriendRecommend程序实现11.4.3建立FirstJob类文件，编写FirstJob类

for

(int

i

=

0;

i

<

friends.length;

i++)

{

/*定义变量从数组中取值*/

String

f1

=

friends[i];

/*返回比较后的数据*/

Fof

ofof

=new

Fof(user，

f1);

/*将数据写入context*/

context.write(ofof，

new

IntWritable(0));

/*遍历数据*/

for

(int

j

=

i+1;

j

<

friends.length;

j++)

{

/*定义变量从数组中取值*/

String

f2

=

friends[j];

/*返回比较后的数据*/

Fof

fof

=new

Fof(f1，

f2);

/*将数据写入context*/

context.write(fof，

new

IntWritable(1));

}

}

}

}

static

class

FofReducer

extends

Reducer<Fof，

IntWritable，

Fof，

IntWritable>

{

protected

void

reduce(Fof

arg0，

Iterable<IntWritable>

arg1，Context

arg2)

throws

IOException，

InterruptedException

{

/*初始化变量*/

int

sum

=0;

boolean

f

=true;

/*遍历数据，计算求和*/

for(IntWritable

i:

arg1){

if(i.get()==0){

f=false;

break;

}else{

sum=sum+i.get();

}

}

if(f){

/*将数据写入context*/

arg2.write(arg0，

new

IntWritable(sum));

}

}

}

}

11.4FriendRecommend程序实现11.4.4建立SecondJob类文件，编写SecondJob类

package

experiment;

import

org.apache.hadoop.io.Text;

import

org.apache.hadoop.io.WritableComparable;

import

org.apache.hadoop.io.WritableComparator;

import

org.apache.hadoop.mapreduce.Mapper;

import

org.apache.hadoop.mapreduce.Reducer;

import

org.apache.hadoop.util.StringUtils;

import

java.io.DataInput;

import

java.io.DataOutput;

import

java.io.IOException;

public

class

SecondJob

{

static

class

SortMapper

extends

Mapper<Text，

Text，

User，

User>

{

protected

void

map(Text

key，

Text

value，

Context

context)

throws

IOException，

InterruptedException

{

/*获取value的值并进行数据分割*/

String[]

args=StringUtils.split(value.toString()，'\t');

/*定义变量并赋值*/

String

other=args[0];

/*数据类型转换*/

int

friendsCount

=Integer.parseInt(args[1]);

/*将数据写入context*/

context.write(new

User(key.toString()，friendsCount)，

new

User(other，friendsCount));

context.write(new

User(other，friendsCount)，

new

User(key.toString()，friendsCount));

}

}

static

class

SortReducer

extends

Reducer<User，

User，

Text，

Text>

{

protected

void

reduce(User

arg0，

Iterable<User>

arg1，

Context

arg2)

throws

IOException，

InterruptedException

{

/*定义变量并赋值*/

String

user

=arg0.getUname();

StringBuffer

sb

=new

StringBuffer();

/*遍历数据进行字符串拼接*/

for(User

u:

arg1

){

sb.append(u.getUname()+":"+u.getFriendsCount());

sb.append("，");

}

11.4FriendRecommend程序实现11.4.4建立SecondJob类文件，编写SecondJob类

/*键值对数据写入*/

arg2.write(new

Text(user)，

new

Text(sb.toString()));

}

}

public

static

class

User

implements

WritableComparable<User>

{

private

String

uname;

private

int

friendsCount;

public

String

getUname()

{

return

uname;

}

public

void

setUname(String

uname)

{

this.uname

=

uname;

}

public

int

getFriendsCount()

{

return

friendsCount;

}

public

void

setFriendsCount(int

friendsCount)

{

this.friendsCount

=

friendsCount;

}

public

User()

{

}

public

User(String

uname，

int

friendsCount)

{

this.uname

=

uname;

this.friendsCount

=

friendsCount;

}

/*数据序列化*/

public

void

write(DataOutput

out)

throws

IOException

{

out.writeUTF(uname);

out.writeInt(friendsCount);

}

/*数据反序列换*/

public

void

readFields(DataInput

in)

throws

IOException

{

this.uname

=

in.readUTF();

this.friendsCount

=

in.readInt();

}

11.4FriendRecommend程序实现11.4.4建立SecondJob类文件，编写SecondJob类

/*数据对比*/

public

int

compareTo(User

o)

{

int

result

=

this.pareTo(o.getUname());

if

(result

==

0)

{

return

Ipare(this.friendsCount，

o.getFriendsCount());

}

return

result;

}

}

public

static

class

FoFSort

extends

WritableComparator

{

public

FoFSort()

{

super(User.class，true);

}

/*数据进行比较排序*/

public

int

compare(WritableComparable

a，

WritableComparable

b)

{

User

u1

=(User)

a;

User

u2=(User)

b;

int

result

=u1.getUname().compareTo(u2.getUname());

if(result==0){

return

-Ipare(u1.getFriendsCount()，

u2.getFriendsCount());

}

return

result;

}

}

public

static

class

FoFGroup

extends

WritableComparator{

public

FoFGroup()

{

super(User.class，true);

}

/*数据比较*/

public

int

compare(WritableComparable

a，

WritableComparable

b)

{

User

u1

=(User)

a;

User

u2=(User)

b;

return

u1.getUname().compareTo(u2.getUname());

}

}

}

11.4FriendRecommend程序实现11.4.5建立Run类文件，编写Run类。packageexperiment;importorg.apache.hadoop.conf.Configuration;importorg.apache.hadoop.fs.FileSystem;importorg.apache.hadoop.fs.Path;importorg.apache.hadoop.io.IntWritable;importorg.apache.hadoop.mapreduce.Job;importorg.apache.hadoop.mapreduce.lib.input.FileInputFormat;importorg.apache.hadoop.mapreduce.lib.input.KeyValueTextInputFormat;importorg.apache.hadoop.mapreduce.lib.output.FileOutputFormat;publicclassRun{publicstaticvoidmain(String[]args){/*创建配置对象并设置属性*/Configurationconfig=newConfiguration();config.set("fs.defaultFS"，"hdfs://master:9000");/*作业1成功运行后，运行作业2*/if(run1(config)){run2(config);}

}publicstaticbooleanrun1(Configurationconfig){try{/*获取文件系统*/FileSystemfs=FileSystem.get(config);/*获取作业*/Jobjob=Job.getInstance(config);/*设置jar包源*/job.setJarByClass(Run.class);/*设置作业名称*/job.setJobName("friend");/*指定作业Mapper类*/job.setMapperClass(FirstJob.FofMapper.class);/*指定作业Reducer类*/job.setReducerClass(FirstJob.FofReducer.class);/*设置作业输出数据的键类*/job.setMapOutputKeyClass(FirstJob.Fof.class);/*设置作业输出数据的值类*/job.setMapOutputValueClass(IntWritable.class);/*设置作业的输入格式类*/job.setInputFormatClass(KeyValueTextInputFormat.class);

11.4FriendRecommend程序实现11.4.5建立Run类文件，编写Run类。publicstaticvoidrun2(Configurationconfig){try{/*获取文件系统*/FileSystemfs=FileSystem.get(config);/*获取作业*/Jobjob=Job.getInstance(config);/*设置jar包源*/job.setJarByClass(Run.class);/*设置作业名称*/job.setJobName("fof2");/*指定作业Mapper类*/job.setMapperClass(SecondJob.SortMapper.class);/*指定作业Reducer类*/job.setReducerClass(SecondJob.SortReducer.class);/*指定作业排序*/job.setSortComparatorClass(SecondJob.FoFSort.class);job.setGroupingComparatorClass(SecondJob.FoFGroup.class);/*设置作业输出数据的键类*/job.setMapOutputKeyClass(SecondJob.User.class);/*设置作业输出数据的值类*/job.setMapOutputValueClass(SecondJob.User.class);/*设置作业输入路径*/FileInputFormat.addInputPath(job，newPath("/root/experiment/datas/"));/*定义输出路径*/Pathoutpath=newPath("/root/experiment/output/output1");/*路径存在则删除路径*/if(fs.exists(outpath)){fs.delete(outpath，true);}/*设置作业输出路径*/FileOutputFormat.setOutputPath(job，outpath);/*提交作业获取返回值