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

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

了解好友推薦項目需求和目標

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

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

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

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的值并進行數(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ù)進行字符串拼接*/

for(User

u:

arg1

){

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

sb.append("，");

}

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

/*鍵值對數(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ù)對比*/

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ù)進行比較排序*/

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)建配置對象并設(shè)置屬性*/Configurationconfig=newConfiguration();config.set("fs.defaultFS"，"hdfs://master:9000");/*作業(yè)1成功運行后，運行作業(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è)獲取返回值