大數(shù)據(jù)技術(shù)基礎(chǔ)教程 課件 第11章 好友案例分析

第11章好友推薦案例分析《大數(shù)據(jù)技術(shù)基礎(chǔ)教程》學(xué)習(xí)目標(biāo)/Target

了解好友推薦項(xiàng)目需求和目標(biāo)

掌握項(xiàng)目所需的預(yù)備知識(shí)實(shí)踐如何編寫MapReduce程序，完成QQ好友推薦操作章節(jié)概述/

Summary近十幾年間，隨著在線社交網(wǎng)絡(luò)的蓬勃發(fā)展，研究人員才開始有機(jī)會(huì)在大量現(xiàn)實(shí)數(shù)據(jù)的基礎(chǔ)上對(duì)社交影響力進(jìn)行建模和分析，并取得了豐碩的研究成果和廣泛的應(yīng)用價(jià)值。在這上千萬用戶的社交群體里，尋找潛在好友進(jìn)行推薦，成為社交網(wǎng)絡(luò)分析的關(guān)鍵問題之一。社交網(wǎng)絡(luò)中的好友是指在社交網(wǎng)絡(luò)中出現(xiàn)在用戶聯(lián)系人列表中的用戶，它是一種廣義上的朋友，既可以是Facebook中的好友，也可是新浪微博中的關(guān)注用戶等。為社交網(wǎng)絡(luò)用戶推薦好友就是幫助用戶在社交網(wǎng)絡(luò)中找到她們感興趣的用戶，進(jìn)而添加到自己的聯(lián)系人列表之中，方便進(jìn)一步的交流及商業(yè)應(yīng)用。本章主要介紹了如何編寫MapReduce程序，完成QQ好友推薦操作的相關(guān)知識(shí)目錄/Contents01020304任務(wù)需求實(shí)驗(yàn)前HDFS平臺(tái)的準(zhǔn)備工作IDEA下建立Maven項(xiàng)目projectFriendRecommend程序?qū)崿F(xiàn)05運(yùn)行程序與結(jié)果驗(yàn)證任務(wù)需求11.111.1任務(wù)需求通過mapreduce計(jì)算，為每個(gè)QQ用戶推薦好友。數(shù)據(jù)集第一列為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實(shí)驗(yàn)前HDFS平臺(tái)的準(zhǔn)備工作11.211.2實(shí)驗(yàn)前HDFS平臺(tái)的準(zhǔn)備工作11.2.1啟動(dòng)Hadoop平臺(tái)啟動(dòng)Hadoop平臺(tái)11.2實(shí)驗(yàn)前HDFS平臺(tái)的準(zhǔn)備工作11.2.2

實(shí)驗(yàn)前數(shù)據(jù)路徑的準(zhǔn)備11.2實(shí)驗(yàn)前HDFS平臺(tái)的準(zhǔn)備工作11.2.3實(shí)驗(yàn)程序每次執(zhí)行前，運(yùn)行結(jié)果輸出路徑的準(zhǔn)備每次實(shí)驗(yàn)程序運(yùn)行結(jié)果會(huì)存儲(chǔ)在HDFS平臺(tái)的/root/experiment/output/路徑下，所以在程序執(zhí)行前，該路徑不能存在，如果存在，需要進(jìn)行刪除。IDEA下建立Maven項(xiàng)目project11.311.3IDEA下建立Maven項(xiàng)目project1)建立新項(xiàng)目1.打開IDEA工具。2.確定要建立的項(xiàng)目類型11.3IDEA下建立Maven項(xiàng)目project1)建立新項(xiàng)目3.填入工程信息4.確認(rèn)項(xiàng)目信息，完成項(xiàng)目創(chuàng)建11.3IDEA下建立Maven項(xiàng)目project1)建立新項(xiàng)目5.此時(shí)進(jìn)入IDEA的開發(fā)界面。如果在開發(fā)界面的上方彈出“TipoftheDay”窗口，點(diǎn)出”Close”按鈕，關(guān)閉該窗口即可。6.在右下角彈出的對(duì)話框中，選擇EnableAuto-Import（如未彈出該對(duì)話框請(qǐng)忽略此步驟）。7.顯示IDEA開發(fā)環(huán)境的主窗口。11.3IDEA下建立Maven項(xiàng)目project2)配置pom.xml文件11.3IDEA下建立Maven項(xiàng)目project3)查看Hadoop工程的Maven依賴包FriendRecommend程序?qū)崿F(xiàn)11.411.4FriendRecommend程序?qū)崿F(xiàn)11.4.1數(shù)據(jù)準(zhǔn)備11.4FriendRecommend程序?qū)崿F(xiàn)11.4.2編寫單表關(guān)聯(lián)的MapReduce程序11.4FriendRecommend程序?qū)崿F(xiàn)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');

/*遍歷數(shù)據(jù)*/

11.4FriendRecommend程序?qū)崿F(xiàn)11.4.3建立FirstJob類文件，編寫FirstJob類

for

(int

i

=

0;

i

<

friends.length;

i++)

{

/*定義變量從數(shù)組中取值*/

String

f1

=

friends[i];

/*返回比較后的數(shù)據(jù)*/

Fof

ofof

=new

Fof(user，

f1);

/*將數(shù)據(jù)寫入context*/

context.write(ofof，

new

IntWritable(0));

/*遍歷數(shù)據(jù)*/

for

(int

j

=

i+1;

j

<

friends.length;

j++)

{

/*定義變量從數(shù)組中取值*/

String

f2

=

friends[j];

/*返回比較后的數(shù)據(jù)*/

Fof

fof

=new

Fof(f1，

f2);

/*將數(shù)據(jù)寫入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;

/*遍歷數(shù)據(jù)，計(jì)算求和*/

for(IntWritable

i:

arg1){

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

f=false;

break;

}else{

sum=sum+i.get();

}

}

if(f){

/*將數(shù)據(jù)寫入context*/

arg2.write(arg0，

new

IntWritable(sum));

}

}

}

}

11.4FriendRecommend程序?qū)崿F(xiàn)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的值并進(jìn)行數(shù)據(jù)分割*/

String[]

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

/*定義變量并賦值*/

String

other=args[0];

/*數(shù)據(jù)類型轉(zhuǎn)換*/

int

friendsCount

=Integer.parseInt(args[1]);

/*將數(shù)據(jù)寫入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();

/*遍歷數(shù)據(jù)進(jìn)行字符串拼接*/

for(User

u:

arg1

){

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

sb.append("，");

}

11.4FriendRecommend程序?qū)崿F(xiàn)11.4.4建立SecondJob類文件，編寫SecondJob類

/*鍵值對(duì)數(shù)據(jù)寫入*/

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;

}

/*數(shù)據(jù)序列化*/

public

void

write(DataOutput

out)

throws

IOException

{

out.writeUTF(uname);

out.writeInt(friendsCount);

}

/*數(shù)據(jù)反序列換*/

public

void

readFields(DataInput

in)

throws

IOException

{

this.uname

=

in.readUTF();

this.friendsCount

=

in.readInt();

}

11.4FriendRecommend程序?qū)崿F(xiàn)11.4.4建立SecondJob類文件，編寫SecondJob類

/*數(shù)據(jù)對(duì)比*/

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);

}

/*數(shù)據(jù)進(jìn)行比較排序*/

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);

}

/*數(shù)據(jù)比較*/

public

int

compare(WritableComparable

a，

WritableComparable

b)

{

User

u1

=(User)

a;

User

u2=(User)

b;

return

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

}

}

}

11.4FriendRecommend程序?qū)崿F(xiàn)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){/*創(chuàng)建配置對(duì)象并設(shè)置屬性*/Configurationconfig=newConfiguration();config.set("fs.defaultFS"，"hdfs://master:9000");/*作業(yè)1成功運(yùn)行后，運(yùn)行作業(yè)2*/if(run1(config)){run2(config);}

}publicstaticbooleanrun1(Configurationconfig){try{/*獲取文件系統(tǒng)*/FileSystemfs=FileSystem.get(config);/*獲取作業(yè)*/Jobjob=Job.getInstance(config);/*設(shè)置jar包源*/job.setJarByClass(Run.class);/*設(shè)置作業(yè)名稱*/job.setJobName("friend");/*指定作業(yè)Mapper類*/job.setMapperClass(FirstJob.FofMapper.class);/*指定作業(yè)Reducer類*/job.setReducerClass(FirstJob.FofReducer.class);/*設(shè)置作業(yè)輸出數(shù)據(jù)的鍵類*/job.setMapOutputKeyClass(FirstJob.Fof.class);/*設(shè)置作業(yè)輸出數(shù)據(jù)的值類*/job.setMapOutputValueClass(IntWritable.class);/*設(shè)置作業(yè)的輸入格式類*/job.setInputFormatClass(KeyValueTextInputFormat.class);

11.4FriendRecommend程序?qū)崿F(xiàn)11.4.5建立Run類文件，編寫Run類。publicstaticvoidrun2(Configurationconfig){try{/*獲取文件系統(tǒng)*/FileSystemfs=FileSystem.get(config);/*獲取作業(yè)*/Jobjob=Job.getInstance(config);/*設(shè)置jar包源*/job.setJarByClass(Run.class);/*設(shè)置作業(yè)名稱*/job.setJobName("fof2");/*指定作業(yè)Mapper類*/job.setMapperClass(SecondJob.SortMapper.class);/*指定作業(yè)Reducer類*/job.setReducerClass(SecondJob.SortReducer.class);/*指定作業(yè)排序*/job.setSortComparatorClass(SecondJob.FoFSort.class);job.setGroupingComparatorClass(SecondJob.FoFGroup.class);/*設(shè)置作業(yè)輸出數(shù)據(jù)的鍵類*/job.setMapOutputKeyClass(SecondJob.User.class);/*設(shè)置作業(yè)輸出數(shù)據(jù)的值類*/job.setMapOutputValueClass(SecondJob.User.class);/*設(shè)置作業(yè)輸入路徑*/FileInputFormat.addInputPath(job，newPath("/root/experiment/datas/"));/*定義輸出路徑*/Pathoutpath=newPath("/root/experiment/output/output1");/*路徑存在則刪除路徑*/if(fs.exists(outpath)){fs.delete(outpath，true);}/*設(shè)置作業(yè)輸出路徑*/FileOutputFormat.setOutputPath(job，outpath);/*提交作業(yè)獲取返回值