• Posts tagged "rmr"

Blog Archives

RHadoop实验 – 统计邮箱出现次数

RHadoop实践系列文章,包含了R语言与Hadoop结合进行海量数据分析。Hadoop主要用来存储海量数据,R语言完成MapReduce 算法,用来替代Java的MapReduce实现。有了RHadoop可以让广大的R语言爱好者,有更强大的工具处理大数据1G, 10G, 100G, TB, PB。 由于大数据所带来的单机性能问题,可能会一去不复返了。

RHadoop实践是一套系列文章,主要包括”Hadoop环境搭建”,”RHadoop安装与使用”,R实现MapReduce的协同过滤算法”,”HBase和rhbase的安装与使用”。对于单独的R语言爱好者,Java爱好者,或者Hadoop爱好者来说,同时具备三种语言知识并不容 易。此文虽为入门文章,但R,Java,Hadoop基础知识还是需要大家提前掌握。

关于作者:

  • 张丹(Conan), 程序员Java,R,PHP,Javascript
  • weibo:@Conan_Z
  • blog: http://blog.fens.me
  • email: bsspirit@gmail.com

转载请注明出处:
http://blog.fens.me/rhadoop-demo-email/

rhadoop-demo-email

目录

  1. 需求描述
  2. 实验数据
  3. 算法实现

1. 需求描述

基于RHADOOP通过rmr包实现MapReduce算法

  • 1). 计算邮箱域出现了多少次
  • 2). 按次数从大到小排序

例如:
163.com,14
sohu.com,2

2. 实验数据

wolys@21cn.com
zss1984@126.com
294522652@qq.com
simulateboy@163.com
zhoushigang_123@163.com
sirenxing424@126.com
lixinyu23@qq.com
chenlei1201@gmail.com
370433835@qq.com
cxx0409@126.com
viv093@sina.com
q62148830@163.com
65993266@qq.com
summeredison@sohu.com
zhangbao-autumn@163.com
diduo_007@yahoo.com.cn
fxh852@163.com
weiyang1128@163.com
licaijun007@163.com
junhongshouji@126.com
wuxiaohong11111@163.com
fennal@sina.com
li_dao888@163.com
bokil.xu@163.com
362212053@qq.com
youloveyingying@yahoo.cn
boiny@126.com
linlixian200606@126.com
alex126126@126.com
654468252@qq.com
huangdaqiao@yahoo.com.cn
kitty12502@163.com
xl200811@sohu.com
ysjd8@163.com
851627938@qq.com
wubo_1225@163.com
kangtezc@163.com
xiao2018@126.com
121641873@qq.com
296489419@qq.com
beibeilong012@126.com

3. 算法实现

1). 计算邮箱域出现了多少次

把数据上传到HDFS


library(rmr2)
data<-read.table(file="hadoop15.txt")
d0<-to.dfs(keyval(1, data))
from.dfs(d0)

输出:


$key
[1] 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1
[39] 1 1 1
$val
V1
1 wolys@21cn.com
2 zss1984@126.com
3 294522652@qq.com
4 simulateboy@163.com
5 zhoushigang_123@163.com
6 sirenxing424@126.com
7 lixinyu23@qq.com
8 chenlei1201@gmail.com
9 370433835@qq.com
10 cxx0409@126.com
11 viv093@sina.com
12 q62148830@163.com
13 65993266@qq.com
14 summeredison@sohu.com
15 zhangbao-autumn@163.com
16 diduo_007@yahoo.com.cn
17 fxh852@163.com
18 weiyang1128@163.com
19 licaijun007@163.com
20 junhongshouji@126.com
21 wuxiaohong11111@163.com
22 fennal@sina.com
23 li_dao888@163.com
24 bokil.xu@163.com
25 362212053@qq.com
26 youloveyingying@yahoo.cn
27 boiny@126.com
28 linlixian200606@126.com
29 alex126126@126.com
30 654468252@qq.com
31 huangdaqiao@yahoo.com.cn
32 kitty12502@163.com
33 xl200811@sohu.com
34 ysjd8@163.com
35 851627938@qq.com
36 wubo_1225@163.com
37 kangtezc@163.com
38 xiao2018@126.com
39 121641873@qq.com
40 296489419@qq.com
41 beibeilong012@126.com

计算邮箱域出现了多少次


mr<-function(input=d0){
map<-function(k,v){
keyval(word(as.character(v$V1), 2, sep = fixed('@')),1)
}
reduce =function(k, v ) {
keyval(k, sum(v))
}
d1<-mapreduce(input=input,map=map,reduce=reduce,combine=TRUE)
}
d1<-mr(d0)
from.dfs(d1)

输出:

$key
[1] "126.com" "163.com" "21cn.com" "gmail.com" "qq.com"
[6] "sina.com" "sohu.com" "yahoo.cn" "yahoo.com.cn"
$val
[1] 9 14 1 1 9 2 2 1 2

2). 按次数从大到小排序


sort<-function(input=d1){
map<-function(k,v){
keyval(1,data.frame(k,v))
}
reduce<-function(k,v){
v2<-v[order(as.integer(v$v),decreasing=TRUE),]
keyval(1,v2)
}
d2<-mapreduce(input=input,map=map,reduce=reduce,combine=TRUE)
}
d2<-sort(d1)
result<-from.dfs(d2)
result$val

输出:


k v
2 163.com 14
1 126.com 9
5 qq.com 9
6 sina.com 2
7 sohu.com 2
9 yahoo.com.cn 2
3 21cn.com 1
4 gmail.com 1
8 yahoo.cn 1

转载请注明出处:
http://blog.fens.me/rhadoop-demo-email/

打赏作者

RHadoop实践系列之三 R实现MapReduce的协同过滤算法

RHadoop实践系列文章,包含了R语言与Hadoop结合进行海量数据分析。Hadoop主要用来存储海量数据,R语言完成MapReduce 算法,用来替代Java的MapReduce实现。有了RHadoop可以让广大的R语言爱好者,有更强大的工具处理大数据1G, 10G, 100G, TB, PB。 由于大数据所带来的单机性能问题,可能会一去不复返了。

RHadoop实践是一套系列文章,主要包括”Hadoop环境搭建”,”RHadoop安装与使用”,R实现MapReduce的协同过滤算法”,”HBase和rhbase的安装与使用”。对于单独的R语言爱好者,Java爱好者,或者Hadoop爱好者来说,同时具备三种语言知识并不容 易。此文虽为入门文章,但R,Java,Hadoop基础知识还是需要大家提前掌握。

由于rmr2的对hadoop操作有一些特殊性,代码实现有一定难度。需要深入学习的同学,请多尝试并思考key/value值的设计。

本文难度为中高级。

关于作者:

  • 张丹(Conan), 程序员Java,R,PHP,Javascript
  • weibo:@Conan_Z
  • blog: http://blog.fens.me
  • email: bsspirit@gmail.com

转载请注明出处:
http://blog.fens.me/rhadoop-mapreduce-rmr/

rhadoop-mapreduce-rmr

第三篇 R实现MapReduce的协同过滤算法,分为3个章节。

1.基于物品推荐的协同过滤算法介绍
2.R本地程序实现
3.R基于Hadoop分步式程序实现

每一章节,都会分为”文字说明部分”和”代码部分”,保持文字说明与代码的连贯性。

注:Hadoop环境及RHadoop的环境,请查看同系列前二篇文章,此文将不再介绍。

1. 基于物品推荐的协同过滤算法介绍

文字说明部分:

越来越多的互联网应用,都开始使用推荐算法(协同过滤算法)。根据用户活跃度和物品流行度,可以分为”基于用户的协同过滤算法”和”基于物品的协同过滤算法”。

基于用户的协同过滤算法,是给用户推荐和他兴趣相似的其他用户喜欢的物品。
基于物品的协同过滤算法,是给用户推荐和他之前喜欢的物品相似的物品。
基于物品的协同过滤算法,是目前广泛使用的一种推荐算法,像Netflix, YouTube, Amazon等。

算法主要分为两步:
1. 计算物品之间的相似度
2. 根据物品的相似度和用户的历史行为给用户生成推荐列表

有关算法的细节请参考:”Mahout In Action”和”推荐系统实践”两本书。

为开发方便,我们选择一组很小的测试数据集。

测试数据,来自于”Mahout In Action” P49
原第8行,3,101,2.5 改为 3,101,2.0
每行3个字段,依次是用户ID,物品ID,对物品的评分

代码部分:

在服务上创建测试数据文件small.csv

~ pwd

/root/R

~ vi small.csv

1,101,5.0
1,102,3.0
1,103,2.5
2,101,2.0
2,102,2.5
2,103,5.0
2,104,2.0
3,101,2.0
3,104,4.0
3,105,4.5
3,107,5.0
4,101,5.0
4,103,3.0
4,104,4.5
4,106,4.0
5,101,4.0
5,102,3.0
5,103,2.0
5,104,4.0
5,105,3.5
5,106,4.0

~ ls

small.csv

2. R本地程序实现

首先,通过R语言实现基于物品的协同过滤算法,为和RHadoop实现进行对比。这里我使用”Mahout In Action”书里,第一章第六节介绍的分步式基于物品的协同过滤算法进行实现。Chapter 6: Distributing recommendation computations

算法的思想:
1. 建立物品的同现矩阵
2. 建立用户对物品的评分矩阵
3. 矩阵计算推荐结果

文字说明部分:

1. 建立物品的同现矩阵

按用户分组,找到每个用户所选的物品,单独出现计数,及两两一组计数。

例如:用户ID为3的用户,分别给101,104,105,107,这4个物品打分。
1) (101,101),(104,104),(105,105),(107,107),单独出现计算各加1。
2) (101,104),(101,105),(101,107),(104,105),(104,107),(105,107),两个一组计数各加1。
3) 把所有用户的计算结果求和,生成一个三角矩阵,再补全三角矩阵,就建立了物品的同现矩阵。

如下面矩阵所示:


      [101] [102] [103] [104] [105] [106] [107]
[101]   5     3     4     4     2     2     1
[102]   3     3     3     2     1     1     0
[103]   4     3     4     3     1     2     0
[104]   4     2     3     4     2     2     1
[105]   2     1     1     2     2     1     1
[106]   2     1     2     2     1     2     0
[107]   1     0     0     1     1     0     1

2. 建立用户对物品的评分矩阵

按用户分组,找到每个用户所选的物品及评分

例如:用户ID为3的用户,分别给(3,101,2.0),(3,104,4.0),(3,105,4.5),(3,107,5.0),这4个物品打分。
1) 找到物品评分(3,101,2.0),(3,104,4.0),(3,105,4.5),(3,107,5.0)
2) 建立用户对物品的评分矩阵


       U3
[101] 2.0
[102] 0.0
[103] 0.0
[104] 4.0
[105] 4.5
[106] 0.0
[107] 5.0

3. 矩阵计算推荐结果

同现矩阵*评分矩阵=推荐结果

alogrithm_1

图片摘自”Mahout In Action”

推荐给用户ID为3的用户的结果是(103,24.5),(102,18.5),(106,16.5)

代码部分:


#引用plyr包
library(plyr)

#读取数据集
train<-read.csv(file="small.csv",header=FALSE)
names(train)<-c("user","item","pref") 

> train
  user item pref
1 1 101 5.0
2 1 102 3.0
3 1 103 2.5
4 2 101 2.0
5 2 102 2.5
6 2 103 5.0
7 2 104 2.0
8 3 101 2.0
9 3 104 4.0
10 3 105 4.5
11 3 107 5.0
12 4 101 5.0
13 4 103 3.0
14 4 104 4.5
15 4 106 4.0
16 5 101 4.0
17 5 102 3.0
18 5 103 2.0
19 5 104 4.0
20 5 105 3.5
21 5 106 4.0

#计算用户列表
usersUnique<-function(){
  users<-unique(train$user)
  users[order(users)]
}

#计算商品列表方法
itemsUnique<-function(){
  items<-unique(train$item)
  items[order(items)]
}

# 用户列表
users<-usersUnique() 
> users
[1] 1 2 3 4 5

# 商品列表
items<-itemsUnique() 
> items
[1] 101 102 103 104 105 106 107

#建立商品列表索引
index<-function(x) which(items %in% x)
data<-ddply(train,.(user,item,pref),summarize,idx=index(item)) 

> data
 user item pref idx
1 1 101 5.0 1
2 1 102 3.0 2
3 1 103 2.5 3
4 2 101 2.0 1
5 2 102 2.5 2
6 2 103 5.0 3
7 2 104 2.0 4
8 3 101 2.0 1
9 3 104 4.0 4
10 3 105 4.5 5
11 3 107 5.0 7
12 4 101 5.0 1
13 4 103 3.0 3
14 4 104 4.5 4
15 4 106 4.0 6
16 5 101 4.0 1
17 5 102 3.0 2
18 5 103 2.0 3
19 5 104 4.0 4
20 5 105 3.5 5
21 5 106 4.0 6

#同现矩阵
cooccurrence<-function(data){
  n<-length(items)
  co<-matrix(rep(0,n*n),nrow=n)
  for(u in users){
    idx<-index(data$item[which(data$user==u)])
    m<-merge(idx,idx)
    for(i in 1:nrow(m)){
      co[m$x[i],m$y[i]]=co[m$x[i],m$y[i]]+1
    }
  }
  return(co)
}

#推荐算法
recommend<-function(udata=udata,co=coMatrix,num=0){
  n<-length(items)

  # all of pref
  pref<-rep(0,n)
  pref[udata$idx]<-udata$pref

  # 用户评分矩阵
  userx<-matrix(pref,nrow=n)

  # 同现矩阵*评分矩阵
  r<-co %*% userx

  # 推荐结果排序
   r[udata$idx]<-0
  idx<-order(r,decreasing=TRUE)
  topn<-data.frame(user=rep(udata$user[1],length(idx)),item=items[idx],val=r[idx])
  topn<-topn[which(topn$val>0),]

  # 推荐结果取前num个
  if(num>0){
    topn<-head(topn,num)
  }

  #返回结果
  return(topn)
}

#生成同现矩阵
co<-cooccurrence(data) 
> co
    [,1] [,2] [,3] [,4] [,5] [,6] [,7]
[1,]  5    3    4    4    2    2    1
[2,]  3    3    3    2    1    1    0
[3,]  4    3    4    3    1    2    0
[4,]  4    2    3    4    2    2    1
[5,]  2    1    1    2    2    1    1
[6,]  2    1    2    2    1    2    0
[7,]  1    0    0    1    1    0    1

#计算推荐结果
recommendation<-data.frame()
for(i in 1:length(users)){
  udata<-data[which(data$user==users[i]),]
  recommendation<-rbind(recommendation,recommend(udata,co,0)) 
} 

> recommendation
  user item val
1 1 104 33.5
2 1 106 18.0
3 1 105 15.5
4 1 107 5.0
5 2 106 20.5
6 2 105 15.5
7 2 107 4.0
8 3 103 24.5
9 3 102 18.5
10 3 106 16.5
11 4 102 37.0
12 4 105 26.0
13 4 107 9.5
14 5 107 11.5

3. R基于Hadoop分步式程序实现

R语言实现的MapReduce算法,可以基于R的数据对象实现,不必如JAVA一样使用文本存储。

算法思想同上面R语言实现思想,略有复杂。

算法的思想:
1. 建立物品的同现矩阵
1) 按用户分组,得到所有物品出现的组合列表。
2) 对物品组合列表进行计数,建立物品的同现矩阵
2. 建立用户对物品的评分矩阵
3. 合并同现矩阵和评分矩阵
4. 计算推荐结果列表
5. 按输入格式得到推荐评分列表

通过MapReduce实现时,所有操作都要使用Map和Reduce的任务完成,程序实现过程略有变化。

aglorithm_2

图片摘自”Mahout In Action”

文字说明部分:

1. 建立物品的同现矩阵

1) 按用户分组,得到所有物品出现的组合列表。

key:物品列表向量
val:物品组合向量


$key
[1] 101 101 101 101 101 101 101 101 101 101 101 101 101 101 101 102 102 102 102
[20] 102 102 102 103 103 103 103 103 103 103 103 103 103 103 104 104 104 104 104
[39] 104 104 104 104 104 104 104 105 105 105 105 106 106 106 106 107 107 107 107
[58] 101 101 101 101 101 101 102 102 102 102 102 102 103 103 103 103 103 103 104
[77] 104 104 104 104 104 105 105 105 105 105 105 106 106 106 106 106 106

$val
[1] 101 102 103 101 102 103 104 101 104 105 107 101 103 104 106 101 102 103 101
[20] 102 103 104 101 102 103 101 102 103 104 101 103 104 106 101 102 103 104 101
[39] 104 105 107 101 103 104 106 101 104 105 107 101 103 104 106 101 104 105 107
[58] 101 102 103 104 105 106 101 102 103 104 105 106 101 102 103 104 105 106 101
[77] 102 103 104 105 106 101 102 103 104 105 106 101 102 103 104 105 106

2) 对物品组合列表进行计数,建立物品的同现矩阵

key:物品列表向量
val:同现矩阵的数据框值(item,item,Freq)
矩阵格式,要与”2. 建立用户对物品的评分矩阵”的格式一致,把异构的两种数据源,合并为同一种数据格式,为”3. 合并 同现矩阵 和 评分矩阵”做数据基础。


$key
[1] 101 101 101 101 101 101 101 102 102 102 102 102 102 103 103 103 103 103 103
[20] 104 104 104 104 104 104 104 105 105 105 105 105 105 105 106 106 106 106 106
[39] 106 107 107 107 107

$val
k v freq
1 101 101 5
2 101 102 3
3 101 103 4
4 101 104 4
5 101 105 2
6 101 106 2
7 101 107 1
8 102 101 3
9 102 102 3
10 102 103 3
11 102 104 2
12 102 105 1
13 102 106 1
14 103 101 4
15 103 102 3
16 103 103 4
17 103 104 3
18 103 105 1
19 103 106 2
20 104 101 4
21 104 102 2
22 104 103 3
23 104 104 4
24 104 105 2
25 104 106 2
26 104 107 1
27 105 101 2
28 105 102 1
29 105 103 1
30 105 104 2
31 105 105 2
32 105 106 1
33 105 107 1
34 106 101 2
35 106 102 1
36 106 103 2
37 106 104 2
38 106 105 1
39 106 106 2
40 107 101 1
41 107 104 1
42 107 105 1
43 107 107 1

2. 建立用户对物品的评分矩阵

key:物品列表
val:用户对物品打分矩阵
矩阵格式,要与”2) 对物品组合列表进行计数,建立物品的同现矩阵”的格式一致,把异构的两种数据源,合并为同一种数据格式,为”3. 合并 同现矩阵 和 评分矩阵”做数据基础


$key
[1] 101 101 101 101 101 102 102 102 103 103 103 103 104 104 104 104 105 105 106
[20] 106 107

$val
item user pref
1 101 1 5.0
2 101 2 2.0
3 101 3 2.0
4 101 4 5.0
5 101 5 4.0
6 102 1 3.0
7 102 2 2.5
8 102 5 3.0
9 103 1 2.5
10 103 2 5.0
11 103 4 3.0
12 103 5 2.0
13 104 2 2.0
14 104 3 4.0
15 104 4 4.5
16 104 5 4.0
17 105 3 4.5
18 105 5 3.5
19 106 4 4.0
20 106 5 4.0
21 107 3 5.0

3. 合并 同现矩阵 和 评分矩阵

这一步操作是MapReduce比较特殊的,因为数据源是两个异构数据源,进行MapReduce的操作。
在之前,我们已经把两种格式合并为一样的。使用equijoin这个rmr2包的函数,进行矩阵合并。
key:NULL
val:合并的数据框


$key
NULL

$val
k.l v.l freq.l item.r user.r pref.r
1 103 101 4 103 1 2.5
2 103 102 3 103 1 2.5
3 103 103 4 103 1 2.5
4 103 104 3 103 1 2.5
5 103 105 1 103 1 2.5
6 103 106 2 103 1 2.5
7 103 101 4 103 2 5.0
8 103 102 3 103 2 5.0
9 103 103 4 103 2 5.0
10 103 104 3 103 2 5.0
11 103 105 1 103 2 5.0
12 103 106 2 103 2 5.0
13 103 101 4 103 4 3.0
....

4. 计算推荐结果列表

把第三步中的矩阵,进行合并计算,得到推荐结果列表
key:物品列表
val:推荐结果数据框


$key
[1] 101 101 101 101 101 101 101 101 101 101 101 101 101 101 101 101 101 101
[19] 101 101 101 101 101 101 101 101 101 101 101 101 101 101 101 101 101 102
[37] 102 102 102 102 102 102 102 102 102 102 102 102 102 102 102 102 102 103
[55] 103 103 103 103 103 103 103 103 103 103 103 103 103 103 103 103 103 103
[73] 103 103 103 103 103 104 104 104 104 104 104 104 104 104 104 104 104 104
[91] 104 104 104 104 104 104 104 104 104 104 104 104 104 104 104 105 105 105
[109] 105 105 105 105 105 105 105 105 105 105 105 106 106 106 106 106 106 106
[127] 106 106 106 106 106 107 107 107 107

$val
k.l v.l user.r v
1 101 101 1 25.0
2 101 101 2 10.0
3 101 101 3 10.0
4 101 101 4 25.0
5 101 101 5 20.0
6 101 102 1 15.0
7 101 102 2 6.0
8 101 102 3 6.0
9 101 102 4 15.0
10 101 102 5 12.0
11 101 103 1 20.0
12 101 103 2 8.0
13 101 103 3 8.0
14 101 103 4 20.0
15 101 103 5 16.0
16 101 104 1 20.0
17 101 104 2 8.0
18 101 104 3 8.0
....

5. 按输入格式得到推荐评分列表

对推荐结果列表,进行排序处理,输出排序后的推荐结果。
key:用户ID
val:推荐结果数据框


$key
[1] 1 1 1 1 1 1 1 2 2 2 2 2 2 2 3 3 3 3 3 3 3 4 4 4 4 4 4 4 5 5 5 5 5 5 5

$val
user item pref
1 1 101 44.0
2 1 103 39.0
3 1 104 33.5
4 1 102 31.5
5 1 106 18.0
6 1 105 15.5
7 1 107 5.0
8 2 101 45.5
9 2 103 41.5
10 2 104 36.0
11 2 102 32.5
12 2 106 20.5
13 2 105 15.5
14 2 107 4.0
15 3 101 40.0
16 3 104 38.0
17 3 105 26.0
18 3 103 24.5
19 3 102 18.5
20 3 106 16.5
21 3 107 15.5
22 4 101 63.0
23 4 104 55.0
24 4 103 53.5
25 4 102 37.0
26 4 106 33.0
27 4 105 26.0
28 4 107 9.5
29 5 101 68.0
30 5 104 59.0
31 5 103 56.5
32 5 102 42.5
33 5 106 34.5
34 5 105 32.0
35 5 107 11.5

rmr2使用提示:

1) rmr.options(backend = ‘hadoop’)

这里backend有两个值,hadoop,local。hadoop是默认值,使用hadoop环境运行程序。local是一个本地测试的设置,已经不建议再使用。我在开发时,试过local设置,运行速度非常快,模拟了hadoop的运行环境。但是,local模式下的代码,不能和hadoop模式下完全兼容,变动也比较大,因此不建议大家使用。

2) equijoin(…,outer=c(‘left’))

这里outer包括了4个值,c(“”, “left”, “right”, “full”),非常像数据库中两个表的join操作

3) keyval(k,v)

mapReduce的操作,需要key和valve保存数据。如果直接输出,或者输出的未加key,会有一个警告Converting to.dfs argument to keyval with a NULL key。再上一篇文章中,rmr2的例子中就有类似的情况,请大家注意修改代码。


> to.dfs(1:10)

Warning message:
In to.dfs(1:10) : Converting to.dfs argument to keyval with a NULL key

代码部分:

#加载rmr2包
library(rmr2)

#输入数据文件
train<-read.csv(file="small.csv",header=FALSE)
names(train)<-c("user","item","pref")

#使用rmr的hadoop格式,hadoop是默认设置。
rmr.options(backend = 'hadoop')

#把数据集存入HDFS
train.hdfs = to.dfs(keyval(train$user,train))
from.dfs(train.hdfs)

> from.dfs(train.hdfs)

    13/04/07 14:35:44 INFO util.NativeCodeLoader: Loaded the native-hadoop library
    13/04/07 14:35:44 INFO zlib.ZlibFactory: Successfully loaded & initialized native-zlib library
    13/04/07 14:35:44 INFO compress.CodecPool: Got brand-new decompressor
    $key
     [1] 1 1 1 2 2 2 2 3 3 3 3 4 4 4 4 5 5 5 5 5 5

    $val
       user item pref
    1     1  101  5.0
    2     1  102  3.0
    3     1  103  2.5
    4     2  101  2.0
    5     2  102  2.5
    6     2  103  5.0
    7     2  104  2.0
    8     3  101  2.0
    9     3  104  4.0
    10    3  105  4.5
    11    3  107  5.0
    12    4  101  5.0
    13    4  103  3.0
    14    4  104  4.5
    15    4  106  4.0
    16    5  101  4.0
    17    5  102  3.0
    18    5  103  2.0
    19    5  104  4.0
    20    5  105  3.5
    21    5  106  4.0

#STEP 1, 建立物品的同现矩阵
# 1) 按用户分组,得到所有物品出现的组合列表。
train.mr<-mapreduce(
  train.hdfs, 
  map = function(k, v) {
    keyval(k,v$item)
  }
  ,reduce=function(k,v){
    m<-merge(v,v)
    keyval(m$x,m$y)
  }
)

from.dfs(train.mr)

    $key
     [1] 101 101 101 101 101 101 101 101 101 101 101 101 101 101 101 102 102 102 102
    [20] 102 102 102 103 103 103 103 103 103 103 103 103 103 103 104 104 104 104 104
    [39] 104 104 104 104 104 104 104 105 105 105 105 106 106 106 106 107 107 107 107
    [58] 101 101 101 101 101 101 102 102 102 102 102 102 103 103 103 103 103 103 104
    [77] 104 104 104 104 104 105 105 105 105 105 105 106 106 106 106 106 106

    $val
     [1] 101 102 103 101 102 103 104 101 104 105 107 101 103 104 106 101 102 103 101
    [20] 102 103 104 101 102 103 101 102 103 104 101 103 104 106 101 102 103 104 101
    [39] 104 105 107 101 103 104 106 101 104 105 107 101 103 104 106 101 104 105 107
    [58] 101 102 103 104 105 106 101 102 103 104 105 106 101 102 103 104 105 106 101
    [77] 102 103 104 105 106 101 102 103 104 105 106 101 102 103 104 105 106

# 2) 对物品组合列表进行计数,建立物品的同现矩阵
step2.mr<-mapreduce(
  train.mr,
  map = function(k, v) {
    d<-data.frame(k,v)
    d2<-ddply(d,.(k,v),count)

    key<-d2$k
    val<-d2
    keyval(key,val)
  }
)
from.dfs(step2.mr)

    $key
     [1] 101 101 101 101 101 101 101 102 102 102 102 102 102 103 103 103 103 103 103
    [20] 104 104 104 104 104 104 104 105 105 105 105 105 105 105 106 106 106 106 106
    [39] 106 107 107 107 107

    $val
         k   v freq
    1  101 101    5
    2  101 102    3
    3  101 103    4
    4  101 104    4
    5  101 105    2
    6  101 106    2
    7  101 107    1
    8  102 101    3
    9  102 102    3
    10 102 103    3
    11 102 104    2
    12 102 105    1
    13 102 106    1
    14 103 101    4
    15 103 102    3
    16 103 103    4
    17 103 104    3
    18 103 105    1
    19 103 106    2
    20 104 101    4
    21 104 102    2
    22 104 103    3
    23 104 104    4
    24 104 105    2
    25 104 106    2
    26 104 107    1
    27 105 101    2
    28 105 102    1
    29 105 103    1
    30 105 104    2
    31 105 105    2
    32 105 106    1
    33 105 107    1
    34 106 101    2
    35 106 102    1
    36 106 103    2
    37 106 104    2
    38 106 105    1
    39 106 106    2
    40 107 101    1
    41 107 104    1
    42 107 105    1
    43 107 107    1    

# 2. 建立用户对物品的评分矩阵

train2.mr<-mapreduce(
  train.hdfs, 
  map = function(k, v) {
    #df<-v[which(v$user==3),]
    df<-v
    key<-df$item
    val<-data.frame(item=df$item,user=df$user,pref=df$pref)
    keyval(key,val)
  }
)
from.dfs(train2.mr)

    $key
     [1] 101 101 101 101 101 102 102 102 103 103 103 103 104 104 104 104 105 105 106
    [20] 106 107

    $val
       item user pref
    1   101    1  5.0
    2   101    2  2.0
    3   101    3  2.0
    4   101    4  5.0
    5   101    5  4.0
    6   102    1  3.0
    7   102    2  2.5
    8   102    5  3.0
    9   103    1  2.5
    10  103    2  5.0
    11  103    4  3.0
    12  103    5  2.0
    13  104    2  2.0
    14  104    3  4.0
    15  104    4  4.5
    16  104    5  4.0
    17  105    3  4.5
    18  105    5  3.5
    19  106    4  4.0
    20  106    5  4.0
    21  107    3  5.0

#3. 合并同现矩阵 和 评分矩阵
eq.hdfs<-equijoin(
  left.input=step2.mr, 
  right.input=train2.mr,
  map.left=function(k,v){
    keyval(k,v)
  },
  map.right=function(k,v){
    keyval(k,v)
  },
  outer = c("left")
)
from.dfs(eq.hdfs)

    $key
    NULL

    $val
        k.l v.l freq.l item.r user.r pref.r
    1   103 101      4    103      1    2.5
    2   103 102      3    103      1    2.5
    3   103 103      4    103      1    2.5
    4   103 104      3    103      1    2.5
    5   103 105      1    103      1    2.5
    6   103 106      2    103      1    2.5
    7   103 101      4    103      2    5.0
    8   103 102      3    103      2    5.0
    9   103 103      4    103      2    5.0
    10  103 104      3    103      2    5.0
    11  103 105      1    103      2    5.0
    12  103 106      2    103      2    5.0
    13  103 101      4    103      4    3.0
    14  103 102      3    103      4    3.0
    15  103 103      4    103      4    3.0
    16  103 104      3    103      4    3.0
    17  103 105      1    103      4    3.0
    18  103 106      2    103      4    3.0
    19  103 101      4    103      5    2.0
    20  103 102      3    103      5    2.0
    21  103 103      4    103      5    2.0
    22  103 104      3    103      5    2.0
    23  103 105      1    103      5    2.0
    24  103 106      2    103      5    2.0
    25  101 101      5    101      1    5.0
    26  101 102      3    101      1    5.0
    27  101 103      4    101      1    5.0
    28  101 104      4    101      1    5.0
    29  101 105      2    101      1    5.0
    30  101 106      2    101      1    5.0
    31  101 107      1    101      1    5.0
    32  101 101      5    101      2    2.0
    33  101 102      3    101      2    2.0
    34  101 103      4    101      2    2.0
    35  101 104      4    101      2    2.0
    36  101 105      2    101      2    2.0
    37  101 106      2    101      2    2.0
    38  101 107      1    101      2    2.0
    39  101 101      5    101      3    2.0
    40  101 102      3    101      3    2.0
    41  101 103      4    101      3    2.0
    42  101 104      4    101      3    2.0
    43  101 105      2    101      3    2.0
    44  101 106      2    101      3    2.0
    45  101 107      1    101      3    2.0
    46  101 101      5    101      4    5.0
    47  101 102      3    101      4    5.0
    48  101 103      4    101      4    5.0
    49  101 104      4    101      4    5.0
    50  101 105      2    101      4    5.0
    51  101 106      2    101      4    5.0
    52  101 107      1    101      4    5.0
    53  101 101      5    101      5    4.0
    54  101 102      3    101      5    4.0
    55  101 103      4    101      5    4.0
    56  101 104      4    101      5    4.0
    57  101 105      2    101      5    4.0
    58  101 106      2    101      5    4.0
    59  101 107      1    101      5    4.0
    60  105 101      2    105      3    4.5
    61  105 102      1    105      3    4.5
    62  105 103      1    105      3    4.5
    63  105 104      2    105      3    4.5
    64  105 105      2    105      3    4.5
    65  105 106      1    105      3    4.5
    66  105 107      1    105      3    4.5
    67  105 101      2    105      5    3.5
    68  105 102      1    105      5    3.5
    69  105 103      1    105      5    3.5
    70  105 104      2    105      5    3.5
    71  105 105      2    105      5    3.5
    72  105 106      1    105      5    3.5
    73  105 107      1    105      5    3.5
    74  106 101      2    106      4    4.0
    75  106 102      1    106      4    4.0
    76  106 103      2    106      4    4.0
    77  106 104      2    106      4    4.0
    78  106 105      1    106      4    4.0
    79  106 106      2    106      4    4.0
    80  106 101      2    106      5    4.0
    81  106 102      1    106      5    4.0
    82  106 103      2    106      5    4.0
    83  106 104      2    106      5    4.0
    84  106 105      1    106      5    4.0
    85  106 106      2    106      5    4.0
    86  104 101      4    104      2    2.0
    87  104 102      2    104      2    2.0
    88  104 103      3    104      2    2.0
    89  104 104      4    104      2    2.0
    90  104 105      2    104      2    2.0
    91  104 106      2    104      2    2.0
    92  104 107      1    104      2    2.0
    93  104 101      4    104      3    4.0
    94  104 102      2    104      3    4.0
    95  104 103      3    104      3    4.0
    96  104 104      4    104      3    4.0
    97  104 105      2    104      3    4.0
    98  104 106      2    104      3    4.0
    99  104 107      1    104      3    4.0
    100 104 101      4    104      4    4.5
    101 104 102      2    104      4    4.5
    102 104 103      3    104      4    4.5
    103 104 104      4    104      4    4.5
    104 104 105      2    104      4    4.5
    105 104 106      2    104      4    4.5
    106 104 107      1    104      4    4.5
    107 104 101      4    104      5    4.0
    108 104 102      2    104      5    4.0
    109 104 103      3    104      5    4.0
    110 104 104      4    104      5    4.0
    111 104 105      2    104      5    4.0
    112 104 106      2    104      5    4.0
    113 104 107      1    104      5    4.0
    114 102 101      3    102      1    3.0
    115 102 102      3    102      1    3.0
    116 102 103      3    102      1    3.0
    117 102 104      2    102      1    3.0
    118 102 105      1    102      1    3.0
    119 102 106      1    102      1    3.0
    120 102 101      3    102      2    2.5
    121 102 102      3    102      2    2.5
    122 102 103      3    102      2    2.5
    123 102 104      2    102      2    2.5
    124 102 105      1    102      2    2.5
    125 102 106      1    102      2    2.5
    126 102 101      3    102      5    3.0
    127 102 102      3    102      5    3.0
    128 102 103      3    102      5    3.0
    129 102 104      2    102      5    3.0
    130 102 105      1    102      5    3.0
    131 102 106      1    102      5    3.0
    132 107 101      1    107      3    5.0
    133 107 104      1    107      3    5.0
    134 107 105      1    107      3    5.0
    135 107 107      1    107      3    5.0

#4. 计算推荐结果列表
cal.mr<-mapreduce(
  input=eq.hdfs,
  map=function(k,v){
    val<-v
    na<-is.na(v$user.r)
    if(length(which(na))>0) val<-v[-which(is.na(v$user.r)),]
    keyval(val$k.l,val)
  }
  ,reduce=function(k,v){
    val<-ddply(v,.(k.l,v.l,user.r),summarize,v=freq.l*pref.r)
    keyval(val$k.l,val)
  }
)
from.dfs(cal.mr)

    $key
      [1] 101 101 101 101 101 101 101 101 101 101 101 101 101 101 101 101 101 101
     [19] 101 101 101 101 101 101 101 101 101 101 101 101 101 101 101 101 101 102
     [37] 102 102 102 102 102 102 102 102 102 102 102 102 102 102 102 102 102 103
     [55] 103 103 103 103 103 103 103 103 103 103 103 103 103 103 103 103 103 103
     [73] 103 103 103 103 103 104 104 104 104 104 104 104 104 104 104 104 104 104
     [91] 104 104 104 104 104 104 104 104 104 104 104 104 104 104 104 105 105 105
    [109] 105 105 105 105 105 105 105 105 105 105 105 106 106 106 106 106 106 106
    [127] 106 106 106 106 106 107 107 107 107

    $val
        k.l v.l user.r    v
    1   101 101      1 25.0
    2   101 101      2 10.0
    3   101 101      3 10.0
    4   101 101      4 25.0
    5   101 101      5 20.0
    6   101 102      1 15.0
    7   101 102      2  6.0
    8   101 102      3  6.0
    9   101 102      4 15.0
    10  101 102      5 12.0
    11  101 103      1 20.0
    12  101 103      2  8.0
    13  101 103      3  8.0
    14  101 103      4 20.0
    15  101 103      5 16.0
    16  101 104      1 20.0
    17  101 104      2  8.0
    18  101 104      3  8.0
    19  101 104      4 20.0
    20  101 104      5 16.0
    21  101 105      1 10.0
    22  101 105      2  4.0
    23  101 105      3  4.0
    24  101 105      4 10.0
    25  101 105      5  8.0
    26  101 106      1 10.0
    27  101 106      2  4.0
    28  101 106      3  4.0
    29  101 106      4 10.0
    30  101 106      5  8.0
    31  101 107      1  5.0
    32  101 107      2  2.0
    33  101 107      3  2.0
    34  101 107      4  5.0
    35  101 107      5  4.0
    36  102 101      1  9.0
    37  102 101      2  7.5
    38  102 101      5  9.0
    39  102 102      1  9.0
    40  102 102      2  7.5
    41  102 102      5  9.0
    42  102 103      1  9.0
    43  102 103      2  7.5
    44  102 103      5  9.0
    45  102 104      1  6.0
    46  102 104      2  5.0
    47  102 104      5  6.0
    48  102 105      1  3.0
    49  102 105      2  2.5
    50  102 105      5  3.0
    51  102 106      1  3.0
    52  102 106      2  2.5
    53  102 106      5  3.0
    54  103 101      1 10.0
    55  103 101      2 20.0
    56  103 101      4 12.0
    57  103 101      5  8.0
    58  103 102      1  7.5
    59  103 102      2 15.0
    60  103 102      4  9.0
    61  103 102      5  6.0
    62  103 103      1 10.0
    63  103 103      2 20.0
    64  103 103      4 12.0
    65  103 103      5  8.0
    66  103 104      1  7.5
    67  103 104      2 15.0
    68  103 104      4  9.0
    69  103 104      5  6.0
    70  103 105      1  2.5
    71  103 105      2  5.0
    72  103 105      4  3.0
    73  103 105      5  2.0
    74  103 106      1  5.0
    75  103 106      2 10.0
    76  103 106      4  6.0
    77  103 106      5  4.0
    78  104 101      2  8.0
    79  104 101      3 16.0
    80  104 101      4 18.0
    81  104 101      5 16.0
    82  104 102      2  4.0
    83  104 102      3  8.0
    84  104 102      4  9.0
    85  104 102      5  8.0
    86  104 103      2  6.0
    87  104 103      3 12.0
    88  104 103      4 13.5
    89  104 103      5 12.0
    90  104 104      2  8.0
    91  104 104      3 16.0
    92  104 104      4 18.0
    93  104 104      5 16.0
    94  104 105      2  4.0
    95  104 105      3  8.0
    96  104 105      4  9.0
    97  104 105      5  8.0
    98  104 106      2  4.0
    99  104 106      3  8.0
    100 104 106      4  9.0
    101 104 106      5  8.0
    102 104 107      2  2.0
    103 104 107      3  4.0
    104 104 107      4  4.5
    105 104 107      5  4.0
    106 105 101      3  9.0
    107 105 101      5  7.0
    108 105 102      3  4.5
    109 105 102      5  3.5
    110 105 103      3  4.5
    111 105 103      5  3.5
    112 105 104      3  9.0
    113 105 104      5  7.0
    114 105 105      3  9.0
    115 105 105      5  7.0
    116 105 106      3  4.5
    117 105 106      5  3.5
    118 105 107      3  4.5
    119 105 107      5  3.5
    120 106 101      4  8.0
    121 106 101      5  8.0
    122 106 102      4  4.0
    123 106 102      5  4.0
    124 106 103      4  8.0
    125 106 103      5  8.0
    126 106 104      4  8.0
    127 106 104      5  8.0
    128 106 105      4  4.0
    129 106 105      5  4.0
    130 106 106      4  8.0
    131 106 106      5  8.0
    132 107 101      3  5.0
    133 107 104      3  5.0
    134 107 105      3  5.0
    135 107 107      3  5.0

#5. 按输入格式得到推荐评分列表
result.mr<-mapreduce(
  input=cal.mr,
  map=function(k,v){
    keyval(v$user.r,v)
  }
  ,reduce=function(k,v){
    val<-ddply(v,.(user.r,v.l),summarize,v=sum(v))
    val2<-val[order(val$v,decreasing=TRUE),]
    names(val2)<-c("user","item","pref")
    keyval(val2$user,val2)
  }
)
from.dfs(result.mr)

    $key
     [1] 1 1 1 1 1 1 1 2 2 2 2 2 2 2 3 3 3 3 3 3 3 4 4 4 4 4 4 4 5 5 5 5 5 5 5

    $val
       user item pref
    1     1  101 44.0
    2     1  103 39.0
    3     1  104 33.5
    4     1  102 31.5
    5     1  106 18.0
    6     1  105 15.5
    7     1  107  5.0
    8     2  101 45.5
    9     2  103 41.5
    10    2  104 36.0
    11    2  102 32.5
    12    2  106 20.5
    13    2  105 15.5
    14    2  107  4.0
    15    3  101 40.0
    16    3  104 38.0
    17    3  105 26.0
    18    3  103 24.5
    19    3  102 18.5
    20    3  106 16.5
    21    3  107 15.5
    22    4  101 63.0
    23    4  104 55.0
    24    4  103 53.5
    25    4  102 37.0
    26    4  106 33.0
    27    4  105 26.0
    28    4  107  9.5
    29    5  101 68.0
    30    5  104 59.0
    31    5  103 56.5
    32    5  102 42.5
    33    5  106 34.5
    34    5  105 32.0
    35    5  107 11.5

文章中提供了R用MapReduce方法,实现协同过滤算法的一种思路。

算法可能不是最优的,希望大家有时间写出更好的算法来!随着R语言及Hadoop的发展,相信会有越来越多的算法应用会使用这种方式!

如有问题请给我留言,我很高兴与大家讨论。

######################################################
看文字不过瘾,作者视频讲解,请访问网站:http://onbook.me/video
######################################################

转载请注明出处:
http://blog.fens.me/rhadoop-mapreduce-rmr/

打赏作者

RHadoop实践系列之二:RHadoop安装与使用

RHadoop实践系列文章,包含了R语言与Hadoop结合进行海量数据分析。Hadoop主要用来存储海量数据,R语言完成MapReduce 算法,用来替代Java的MapReduce实现。有了RHadoop可以让广大的R语言爱好者,有更强大的工具处理大数据1G, 10G, 100G, TB, PB。 由于大数据所带来的单机性能问题,可能会一去不复返了。

RHadoop实践是一套系列文章,主要包括”Hadoop环境搭建”,”RHadoop安装与使用”,R实现MapReduce的协同过滤算法”,”HBase和rhbase的安装与使用”。对于单独的R语言爱好者,Java爱好者,或者Hadoop爱好者来说,同时具备三种语言知识并不容 易。此文虽为入门文章,但R,Java,Hadoop基础知识还是需要大家提前掌握。

关于作者:

  • 张丹(Conan), 程序员Java,R,PHP,Javascript
  • weibo:@Conan_Z
  • blog: http://blog.fens.me
  • email: bsspirit@gmail.com

转载请注明出处:
http://blog.fens.me/rhadoop-rhadoop/

rhadoop-rhadoop

第二篇 RHadoop安装与使用部分,分为3个章节。

1. 环境准备
2. RHadoop安装
3. RHadoop程序用例

每一章节,都会分为”文字说明部分”和”代码部分”,保持文字说明与代码的连贯性。

注:Hadoop环境搭建的详细记录,请查看 同系列上一篇文章 “RHadoop实践系列文章之Hadoop环境搭建”。
由于两篇文章并非同一时间所写,hadoop版本及操作系统,分步式环境都略有不同。
两篇文章相互独立,请大家在理解的基础上动手实验,不要完成依赖两篇文章中的运行命令。

环境准备

文字说明部分:

首先环境准备,这里我选择了Linux Ubuntu操作系统12.04的64位版本,大家可以根据自己的使用习惯选择顺手的Linux。

但JDK一定要用Oracle SUN官方的版本,请从官网下载,操作系统的自带的OpenJDK会有各种不兼容。JDK请选择1.6.x的版本,JDK1.7版本也会有各种的不兼容情况。
http://www.oracle.com/technetwork/java/javase/downloads/index.html

Hadoop的环境安装,请参考RHadoop实践系统”Hadoop环境搭建”的一文。

R语言请安装2.15以后的版本,2.14是不能够支持RHadoop的。
如果你也使用Linux Ubuntu操作系统12.04,请先更新软件包源,否则只能下载到2.14版本的R。

代码部分:

1. 操作系统Ubuntu 12.04 x64

~ uname -a
Linux domU-00-16-3e-00-00-85 3.2.0-23-generic #36-Ubuntu SMP Tue Apr 10 20:39:51 UTC 2012 x86_64 x86_64 x86_64 GNU/Linux

2 JAVA环境

~ java -version

java version "1.6.0_29"
Java(TM) SE Runtime Environment (build 1.6.0_29-b11)
Java HotSpot(TM) 64-Bit Server VM (build 20.4-b02, mixed mode)

3 HADOOP环境(这里只需要hadoop)

hadoop-1.0.3  hbase-0.94.2  hive-0.9.0  pig-0.10.0  sqoop-1.4.2  thrift-0.8.0  zookeeper-3.4.4

4 R的环境

R version 2.15.3 (2013-03-01) -- "Security Blanket"
Copyright (C) 2013 The R Foundation for Statistical Computing
ISBN 3-900051-07-0
Platform: x86_64-pc-linux-gnu (64-bit)

4.1 如果是Ubuntu 12.04,请更新源再下载R2.15.3版本

sh -c "echo deb http://mirror.bjtu.edu.cn/cran/bin/linux/ubuntu precise/ >>/etc/apt/sources.list"
apt-get update
apt-get install r-base

RHadoop安装

文字说明部分:

RHadoop是RevolutionAnalytics的工程的项目,开源实现代码在GitHub社区可以找到。RHadoop包含三个R包 (rmr,rhdfs,rhbase),分别是对应Hadoop系统架构中的,MapReduce, HDFS, HBase 三个部分。由于这三个库不能在CRAN中找到,所以需要自己下载。
https://github.com/RevolutionAnalytics/RHadoop/wiki

接下我们需要先安装这三个库的依赖库。
首先是rJava,上个章节我们已经配置好了JDK1.6的环境,运行R CMD javareconf命令,R的程序从系统变量中会读取Java配置。然后打开R程序,通过install.packages的方式,安装rJava。

然后,我还要安装其他的几个依赖库,reshape2,Rcpp,iterators,itertools,digest,RJSONIO,functional,通过install.packages都可以直接安装。

接下安装rhdfs库,在环境变量中增加 HADOOP_CMD 和 HADOOP_STREAMING 两个变量,可以用export在当前命令窗口中增加。但为下次方便使用,最好把变量增加到系统环境变更/etc/environment文件中。再用 R CMD INSTALL安装rhdfs包,就可以顺利完成了。

安装rmr库,使用R CMD INSTALL也可以顺利完成了。

安装rhbase库,后面”HBase和rhbase的安装与使用”文章中会继续介绍,这里暂时跳过。

最后,我们可以查看一下,RHADOOP都安装了哪些库。
由于我的硬盘是外接的,使用mount和软连接(ln -s)挂载了R类库的目录,所以是R的类库在/disk1/system下面
/disk1/system/usr/local/lib/R/site-library/
一般R的类库目录是/usr/lib/R/site-library或者/usr/local/lib/R/site-library,用户也可以使用whereis R的命令查询,自己电脑上R类库的安装位置

代码部分:

1. 下载RHadoop相关的3个程序包

https://github.com/RevolutionAnalytics/RHadoop/wiki/Downloads

rmr-2.1.0
rhdfs-1.0.5
rhbase-1.1

2. 复制到/root/R目录

~/R# pwd
/root/R

~/R# ls
rhbase_1.1.tar.gz  rhdfs_1.0.5.tar.gz  rmr2_2.1.0.tar.gz

3. 安装依赖库

命令行执行
~ R CMD javareconf 
~ R

启动R程序
install.packages("rJava")
install.packages("reshape2")
install.packages("Rcpp")
install.packages("iterators")
install.packages("itertools")
install.packages("digest")
install.packages("RJSONIO")
install.packages("functional")

4. 安装rhdfs库

~ export HADOOP_CMD=/root/hadoop/hadoop-1.0.3/bin/hadoop
~ export HADOOP_STREAMING=/root/hadoop/hadoop-1.0.3/contrib/streaming/hadoop-streaming-1.0.3.jar (rmr2会用到)
~ R CMD INSTALL /root/R/rhdfs_1.0.5.tar.gz 

4.1 最好把HADOOP_CMD设置到环境变量

~ vi /etc/environment

    HADOOP_CMD=/root/hadoop/hadoop-1.0.3/bin/hadoop
    HADOOP_STREAMING=/root/hadoop/hadoop-1.0.3/contrib/streaming/hadoop-streaming-1.0.3.jar

. /etc/environment

5. 安装rmr库

~  R CMD INSTALL rmr2_2.1.0.tar.gz 

6. 安装rhbase库 (暂时跳过)

7. 所有的安装包

~ ls /disk1/system/usr/local/lib/R/site-library/
digest  functional  iterators  itertools  plyr  Rcpp  reshape2  rhdfs  rJava  RJSONIO  rmr2  stringr

RHadoop程序用例

文字说明部分:

安装好rhdfs和rmr两个包后,我们就可以使用R尝试一些hadoop的操作了。

首先,是基本的hdfs的文件操作。

查看hdfs文件目录
hadoop的命令:hadoop fs -ls /user
R语言函数:hdfs.ls(”/user/“)

查看hadoop数据文件
hadoop的命令:hadoop fs -cat /user/hdfs/o_same_school/part-m-00000
R语言函数:hdfs.cat(”/user/hdfs/o_same_school/part-m-00000″)

接下来,我们执行一个rmr算法的任务

普通的R语言程序:

> small.ints = 1:10
> sapply(small.ints, function(x) x^2)

MapReduce的R语言程序:

> small.ints = to.dfs(1:10)
> mapreduce(input = small.ints, map = function(k, v) cbind(v, v^2))
> from.dfs("/tmp/RtmpWnzxl4/file5deb791fcbd5")

因为MapReduce只能访问HDFS文件系统,先要用to.dfs把数据存储到HDFS文件系统里。MapReduce的运算结果再用from.dfs函数从HDFS文件系统中取出。

第二个,rmr的例子是wordcount,对文件中的单词计数

> input<- '/user/hdfs/o_same_school/part-m-00000'
> wordcount = function(input, output = NULL, pattern = " "){

  wc.map = function(., lines) {
            keyval(unlist( strsplit( x = lines,split = pattern)),1)
    }

    wc.reduce =function(word, counts ) {
            keyval(word, sum(counts))
    }         

    mapreduce(input = input ,output = output, input.format = "text",
        map = wc.map, reduce = wc.reduce,combine = T)
}

> wordcount(input)
> from.dfs("/tmp/RtmpfZUFEa/file6cac626aa4a7")

我在HDFS上提前放置了数据文件/user/hdfs/o_same_school/part-m-00000。写wordcount的MapReduce函数,执行wordcount函数,最后用from.dfs从HDFS中取得结果。

代码部分:

1. rhdfs包的使用

启动R程序
> library(rhdfs)

Loading required package: rJava
HADOOP_CMD=/root/hadoop/hadoop-1.0.3/bin/hadoop
Be sure to run hdfs.init()

> hdfs.init()

1.1 命令查看hadoop目录

~ hadoop fs -ls /user

Found 4 items
drwxr-xr-x   - root supergroup          0 2013-02-01 12:15 /user/conan
drwxr-xr-x   - root supergroup          0 2013-03-06 17:24 /user/hdfs
drwxr-xr-x   - root supergroup          0 2013-02-26 16:51 /user/hive
drwxr-xr-x   - root supergroup          0 2013-03-06 17:21 /user/root

1.2 rhdfs查看hadoop目录

> hdfs.ls("/user/")

  permission owner      group size          modtime        file
1 drwxr-xr-x  root supergroup    0 2013-02-01 12:15 /user/conan
2 drwxr-xr-x  root supergroup    0 2013-03-06 17:24  /user/hdfs
3 drwxr-xr-x  root supergroup    0 2013-02-26 16:51  /user/hive
4 drwxr-xr-x  root supergroup    0 2013-03-06 17:21  /user/root

1.3 命令查看hadoop数据文件

~ hadoop fs -cat /user/hdfs/o_same_school/part-m-00000

10,3,tsinghua university,2004-05-26 15:21:00.0
23,4007,北京第一七一中学,2004-05-31 06:51:53.0
51,4016,大连理工大学,2004-05-27 09:38:31.0
89,4017,Amherst College,2004-06-01 16:18:56.0
92,4017,斯坦福大学,2012-11-28 10:33:25.0
99,4017,Stanford University Graduate School of Business,2013-02-19 12:17:15.0
113,4017,Stanford University,2013-02-19 12:17:15.0
123,4019,St Paul's Co-educational College - Hong Kong,2004-05-27 18:04:17.0
138,4019,香港苏浙小学,2004-05-27 18:59:58.0
172,4020,University,2004-05-27 19:14:34.0
182,4026,ff,2004-05-28 04:42:37.0
183,4026,ff,2004-05-28 04:42:37.0
189,4033,tsinghua,2011-09-14 12:00:38.0
195,4035,ba,2004-05-31 07:10:24.0
196,4035,ma,2004-05-31 07:10:24.0
197,4035,southampton university,2013-01-07 15:35:18.0
246,4067,美国史丹佛大学,2004-06-12 10:42:10.0
254,4067,美国史丹佛大学,2004-06-12 10:42:10.0
255,4067,美国休士顿大学,2004-06-12 10:42:10.0
257,4068,清华大学,2004-06-12 10:42:10.0
258,4068,北京八中,2004-06-12 17:34:02.0
262,4068,香港中文大学,2004-06-12 17:34:02.0
310,4070,首都师范大学初等教育学院,2004-06-14 15:35:52.0
312,4070,北京师范大学经济学院,2004-06-14 15:35:52.0

1.4 rhdfs查看hadoop数据文件

>  hdfs.cat("/user/hdfs/o_same_school/part-m-00000")

 [1] "10,3,tsinghua university,2004-05-26 15:21:00.0"
 [2] "23,4007,北京第一七一中学,2004-05-31 06:51:53.0"
 [3] "51,4016,大连理工大学,2004-05-27 09:38:31.0"
 [4] "89,4017,Amherst College,2004-06-01 16:18:56.0"
 [5] "92,4017,斯坦福大学,2012-11-28 10:33:25.0"
 [6] "99,4017,Stanford University Graduate School of Business,2013-02-19 12:17:15.0"
 [7] "113,4017,Stanford University,2013-02-19 12:17:15.0"
 [8] "123,4019,St Paul's Co-educational College - Hong Kong,2004-05-27 18:04:17.0"
 [9] "138,4019,香港苏浙小学,2004-05-27 18:59:58.0"
[10] "172,4020,University,2004-05-27 19:14:34.0"
[11] "182,4026,ff,2004-05-28 04:42:37.0"
[12] "183,4026,ff,2004-05-28 04:42:37.0"
[13] "189,4033,tsinghua,2011-09-14 12:00:38.0"
[14] "195,4035,ba,2004-05-31 07:10:24.0"
[15] "196,4035,ma,2004-05-31 07:10:24.0"
[16] "197,4035,southampton university,2013-01-07 15:35:18.0"
[17] "246,4067,美国史丹佛大学,2004-06-12 10:42:10.0"
[18] "254,4067,美国史丹佛大学,2004-06-12 10:42:10.0"
[19] "255,4067,美国休士顿大学,2004-06-12 10:42:10.0"
[20] "257,4068,清华大学,2004-06-12 10:42:10.0"
[21] "258,4068,北京八中,2004-06-12 17:34:02.0"
[22] "262,4068,香港中文大学,2004-06-12 17:34:02.0"
[23] "310,4070,首都师范大学初等教育学院,2004-06-14 15:35:52.0"
[24] "312,4070,北京师范大学经济学院,2004-06-14 15:35:52.0"

2. rmr2包的使用

启动R程序
> library(rmr2)

Loading required package: Rcpp
Loading required package: RJSONIO
Loading required package: digest
Loading required package: functional
Loading required package: stringr
Loading required package: plyr
Loading required package: reshape2

2.1 执行r任务

> small.ints = 1:10
> sapply(small.ints, function(x) x^2)

[1]   1   4   9  16  25  36  49  64  81 100

2.2 执行rmr2任务

> small.ints = to.dfs(1:10)

13/03/07 12:12:55 INFO util.NativeCodeLoader: Loaded the native-hadoop library
13/03/07 12:12:55 INFO zlib.ZlibFactory: Successfully loaded & initialized native-zlib library
13/03/07 12:12:55 INFO compress.CodecPool: Got brand-new compressor

> mapreduce(input = small.ints, map = function(k, v) cbind(v, v^2))

packageJobJar: [/tmp/RtmpWnzxl4/rmr-local-env5deb2b300d03, /tmp/RtmpWnzxl4/rmr-global-env5deb398a522b, /tmp/RtmpWnzxl4/rmr-streaming-map5deb1552172d, /root/hadoop/tmp/hadoop-unjar7838617732558795635/] [] /tmp/streamjob4380275136001813619.jar tmpDir=null
13/03/07 12:12:59 INFO mapred.FileInputFormat: Total input paths to process : 1
13/03/07 12:12:59 INFO streaming.StreamJob: getLocalDirs(): [/root/hadoop/tmp/mapred/local]
13/03/07 12:12:59 INFO streaming.StreamJob: Running job: job_201302261738_0293
13/03/07 12:12:59 INFO streaming.StreamJob: To kill this job, run:
13/03/07 12:12:59 INFO streaming.StreamJob: /disk1/hadoop/hadoop-1.0.3/libexec/../bin/hadoop job  -Dmapred.job.tracker=hdfs://r.qa.tianji.com:9001 -kill job_201302261738_0293
13/03/07 12:12:59 INFO streaming.StreamJob: Tracking URL: http://192.168.1.243:50030/jobdetails.jsp?jobid=job_201302261738_0293
13/03/07 12:13:00 INFO streaming.StreamJob:  map 0%  reduce 0%
13/03/07 12:13:15 INFO streaming.StreamJob:  map 100%  reduce 0%
13/03/07 12:13:21 INFO streaming.StreamJob:  map 100%  reduce 100%
13/03/07 12:13:21 INFO streaming.StreamJob: Job complete: job_201302261738_0293
13/03/07 12:13:21 INFO streaming.StreamJob: Output: /tmp/RtmpWnzxl4/file5deb791fcbd5

> from.dfs("/tmp/RtmpWnzxl4/file5deb791fcbd5")

$key
NULL

$val
       v
 [1,]  1   1
 [2,]  2   4
 [3,]  3   9
 [4,]  4  16
 [5,]  5  25
 [6,]  6  36
 [7,]  7  49
 [8,]  8  64
 [9,]  9  81
[10,] 10 100

2.3 wordcount执行rmr2任务

> input<- '/user/hdfs/o_same_school/part-m-00000'
> wordcount = function(input, output = NULL, pattern = " "){

    wc.map = function(., lines) {
            keyval(unlist( strsplit( x = lines,split = pattern)),1)
    }

    wc.reduce =function(word, counts ) {
            keyval(word, sum(counts))
    }         

    mapreduce(input = input ,output = output, input.format = "text",
        map = wc.map, reduce = wc.reduce,combine = T)
}

> wordcount(input)

packageJobJar: [/tmp/RtmpfZUFEa/rmr-local-env6cac64020a8f, /tmp/RtmpfZUFEa/rmr-global-env6cac73016df3, /tmp/RtmpfZUFEa/rmr-streaming-map6cac7f145e02, /tmp/RtmpfZUFEa/rmr-streaming-reduce6cac238dbcf, /tmp/RtmpfZUFEa/rmr-streaming-combine6cac2b9098d4, /root/hadoop/tmp/hadoop-unjar6584585621285839347/] [] /tmp/streamjob9195921761644130661.jar tmpDir=null
13/03/07 12:34:41 INFO util.NativeCodeLoader: Loaded the native-hadoop library
13/03/07 12:34:41 WARN snappy.LoadSnappy: Snappy native library not loaded
13/03/07 12:34:41 INFO mapred.FileInputFormat: Total input paths to process : 1
13/03/07 12:34:41 INFO streaming.StreamJob: getLocalDirs(): [/root/hadoop/tmp/mapred/local]
13/03/07 12:34:41 INFO streaming.StreamJob: Running job: job_201302261738_0296
13/03/07 12:34:41 INFO streaming.StreamJob: To kill this job, run:
13/03/07 12:34:41 INFO streaming.StreamJob: /disk1/hadoop/hadoop-1.0.3/libexec/../bin/hadoop job  -Dmapred.job.tracker=hdfs://r.qa.tianji.com:9001 -kill job_201302261738_0296
13/03/07 12:34:41 INFO streaming.StreamJob: Tracking URL: http://192.168.1.243:50030/jobdetails.jsp?jobid=job_201302261738_0296
13/03/07 12:34:42 INFO streaming.StreamJob:  map 0%  reduce 0%
13/03/07 12:34:59 INFO streaming.StreamJob:  map 100%  reduce 0%
13/03/07 12:35:08 INFO streaming.StreamJob:  map 100%  reduce 17%
13/03/07 12:35:14 INFO streaming.StreamJob:  map 100%  reduce 100%
13/03/07 12:35:20 INFO streaming.StreamJob: Job complete: job_201302261738_0296
13/03/07 12:35:20 INFO streaming.StreamJob: Output: /tmp/RtmpfZUFEa/file6cac626aa4a7

> from.dfs("/tmp/RtmpfZUFEa/file6cac626aa4a7")

$key
 [1] "-"
 [2] "04:42:37.0"
 [3] "06:51:53.0"
 [4] "07:10:24.0"
 [5] "09:38:31.0"
 [6] "10:33:25.0"
 [7] "10,3,tsinghua"
 [8] "10:42:10.0"
 [9] "113,4017,Stanford"
[10] "12:00:38.0"
[11] "12:17:15.0"
[12] "123,4019,St"
[13] "138,4019,香港苏浙小学,2004-05-27"
[14] "15:21:00.0"
[15] "15:35:18.0"
[16] "15:35:52.0"
[17] "16:18:56.0"
[18] "172,4020,University,2004-05-27"
[19] "17:34:02.0"
[20] "18:04:17.0"
[21] "182,4026,ff,2004-05-28"
[22] "183,4026,ff,2004-05-28"
[23] "18:59:58.0"
[24] "189,4033,tsinghua,2011-09-14"
[25] "19:14:34.0"
[26] "195,4035,ba,2004-05-31"
[27] "196,4035,ma,2004-05-31"
[28] "197,4035,southampton"
[29] "23,4007,北京第一七一中学,2004-05-31"
[30] "246,4067,美国史丹佛大学,2004-06-12"
[31] "254,4067,美国史丹佛大学,2004-06-12"
[32] "255,4067,美国休士顿大学,2004-06-12"
[33] "257,4068,清华大学,2004-06-12"
[34] "258,4068,北京八中,2004-06-12"
[35] "262,4068,香港中文大学,2004-06-12"
[36] "312,4070,北京师范大学经济学院,2004-06-14"
[37] "51,4016,大连理工大学,2004-05-27"
[38] "89,4017,Amherst"
[39] "92,4017,斯坦福大学,2012-11-28"
[40] "99,4017,Stanford"
[41] "Business,2013-02-19"
[42] "Co-educational"
[43] "College"
[44] "College,2004-06-01"
[45] "Graduate"
[46] "Hong"
[47] "Kong,2004-05-27"
[48] "of"
[49] "Paul's"
[50] "School"
[51] "University"
[52] "university,2004-05-26"
[53] "university,2013-01-07"
[54] "University,2013-02-19"
[55] "310,4070,首都师范大学初等教育学院,2004-06-14"

$val
 [1] 1 2 1 2 1 1 1 4 1 1 2 1 1 1 1 2 1 1 2 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1
[39] 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1

转载请注明出处:
http://blog.fens.me/rhadoop-rhadoop/

打赏作者