Hadoop/R集成I：流处理

如果您平常一直使用MapReduce框架&＃xff0c;那么您可能知道"单词计数示例"是MapReduce的相当于“Hello World&＃xff01;”的一个例子。在之前的帖子中&＃xff0c;我试图稍作改动&＃xff0c;但现在也有一个同样简单的问题 - 按州来计算&＃xff0c;计算房利美(Fannie Mae)地产公司所募集的按揭证券的美元总价的新问题。

到目前为止&＃xff0c;我已经使用了“直接的”Java和Pig&＃xff0c;现在我将注意力转向R(语言).在这篇文章的例子完成之后&＃xff0c;我们将讨论在该情况下R语言的独特之处&＃xff0c;以及为什么字数统计类型的例子不会“真的做正义。在此之前&＃xff0c;我会提及我在这里使用的主要参考资料是Joseph Adler的R in a Nutshell&＃xff08;请参阅第26章&＃xff09;和Tom White的Hadoop&＃xff1a;权威指南&＃xff08;第2章&＃xff09;。

在这里我还是要推荐下我自己建的大数据学习交流qq裙&＃xff1a;522189307 &＃xff0c; 裙 里都是学大数据开发的&＃xff0c;如果你正在学习大数据 &＃xff0c;小编欢迎你加入&＃xff0c;大家都是软件开发党&＃xff0c;不定期分享干货&＃xff08;只有大数据开发相关的&＃xff09;&＃xff0c;包括我自己整理的一份最新的大数据进阶资料和高级开发教程&＃xff0c;欢迎进阶中和进想深入大数据的小伙伴。上述资料加群可以领取

有很多方法可以将R语言与Hadoop结合使用&＃xff0c;其中包括&＃xff1a;

Hadoop流媒体&＃xff0c;这篇文章的主角
RHadoop&＃xff0c;R/Hadoop的集成&＃xff08;请参阅RHadoop Wiki&＃xff09;&＃xff0c;这是将在未来发布的文章的主角。
RHIPE&＃xff08;发音为hree- pay&＃xff09;&＃xff0c;另一个R/Hadoop的集成。
由于我在本博客中试图涵盖的主题十分广泛&＃xff0c;因此我将限制自己使用流式传输和RHadoop。

概览

在Hadoop流中&＃xff0c;您的mapper&＃xff0c;reducer和可选的组合器进程(combiner processes)被写入从标准输入读取并写入标准输出。一旦流程脚本和数据准备就绪&＃xff0c;您只需使用带有一些命令行属性的流式二进制文件就能调用Hadoop。

正如在之前的文章中一样&＃xff0c;我将从房利美的新问题统计库&＃xff08;NIPS&＃xff0c;即New Issue Pool Statistics&＃xff09;文件中获取数据。想要了解更多信息&＃xff0c;请参阅上 一篇文章。我将使用与该文章中相同的数据&＃xff0c;因此我们可以期待结果能够与前面精确匹配。

The Mapper

NIPS文件有一点复杂&＃xff0c;因为它们包含许多不同格式的记录&＃xff08;在 这里查看所有格式&＃xff09;。我们将查看记录类型9的数据来进行我们的分析&＃xff0c;即“GEOGRAPHIC DISTRIBUTION(地理分布&＃xff09;”。我们感兴趣的是第3栏&＃xff08;州名&＃xff09;和第6栏&＃xff08;总的未付余额&＃xff09;。由于在单个文件中混合了多种记录格式&＃xff0c;因此我们首先在管道定界符上将文件分割并丢弃非9类记录。我们需要做的就是输出状态名称和累加未付余额&＃xff0c;每个类型9行包含了1个实例。

我使用RStudio来编写R脚本&＃xff0c;这是一个我通过Coursera上的Roger Peng的Computing for Data Analysis课程了解到的IDE 。在RStudio中进行过交互式脚本构建会话后&＃xff0c;我制作了以下测试脚本&＃xff1a;

#! /usr/bin/env Rscript
conn <- file("/home/hduser/fannie-mae-nips/nips_12262012.txt", open&＃61;"r")
while (length(next.line <- readLines(conn, n&＃61;1)) > 0) {
split.line <- strsplit(next.line, "\\|")
if (as.numeric(split.line[[1]][2]) &＃61;&＃61; 9) {
write(paste(split.line[[1]][3],
gsub("[$,]", "", split.line[[1]][6]), sep&＃61;"\t"), stdout())
}
}
close(conn)
然后我从shell中调用并得到以下输出&＃xff0c;截取了一小段&＃xff1a;

CALIFORNIA 167300.00
FLORIDA 395950.00
GEORGIA 69500.00
ILLINOIS 235200.00
MICHIGAN 781950.00
NEW JERSEY 284550.00
OHIO 334175.00
由于这看起来很干净&＃xff0c;我稍微修改了mapper以生成最终版本&＃xff1a;

#! /usr/bin/env Rscript
conn <- file("stdin", open&＃61;"r")
while (length(next.line <- readLines(conn, n&＃61;1)) > 0) {
split.line <- strsplit(next.line, "\\|")
if (as.numeric(split.line[[1]][2]) &＃61;&＃61; 9) {
write(paste(split.line[[1]][3],
gsub("[$,]", "", split.line[[1]][6]), sep&＃61;"\t"), stdout())
}
}
close(conn)
请注意获取strsplit的结果的下标&＃xff1a;strsplit返回一个列表&＃xff0c;因此文件记录的第2个字段实际上是列表中第一个元素的元素2&＃xff0c;它是解析字段的向量。如果您需要这个结果的详细说明&＃xff0c;请参阅Phil Spector的Data Manipulation with R(使用R的数据操作)的“Subscripting(下标)”一章 _。另外请注意&＃xff0c;gsubto的紧凑型使用可从汇总未付余额中删除美元符号和逗号。

The Reducer

我们的reducer也将从stdin中读取数据&＃xff0c;其中Hadoop运行环境保证了以下内容&＃xff1a;

如果reducer遇到一个关键字&＃xff0c;那么就reducer知道带有该关键字的所有记录都被发送到了该reducer&＃xff0c;所以它可以产生一个输出&＃xff0c;并知道它已经被赋予了该关键字代表的所有记录;
陆续传入的记录会按键排序&＃xff0c;因此reducer知道&＃xff0c;当某个键发生更改时&＃xff0c;未更改之前键的所有记录都已在流中遇到过。
在我们的reducer中&＃xff0c;有两个变量&＃xff1a;一个用于追踪哪个键正在被处理&＃xff0c;另一个用于保存来自给定状态的抵押贷款的总的未支付余额。一旦某个键发生变化&＃xff0c;我们将(使用其键)输出当前的running total出并重置running balance(解释看代码&＃xff0c;running total和runing balance都是作者自己起的名字&＃xff0c;并没有特殊含义)&＃xff1a;

#! /usr/bin/env Rscript
current.key <- NA
current.upb <- 0.0
conn <- file("stdin", open&＃61;"r")
while (length(next.line <- readLines(conn, n&＃61;1)) > 0) {
split.line <- strsplit(next.line, "\t")
key <- split.line[[1]][1]
upb <- as.numeric(split.line[[1]][2])
if (is.na(current.key)) {
current.key <- key
current.upb <- upb
}
else {
if (current.key &＃61;&＃61; key) {
current.upb <- current.upb &＃43; upb
}
else {
write(paste(current.key, current.upb, sep&＃61;"\t"), stdout())
current.key <- key
current.upb <- upb
}
}
}
write(paste(current.key, current.upb, sep&＃61;"\t"), stdout())
close(conn)
现在&＃xff0c;如果我想在单个文件上测试这个reducer&＃xff0c;但是我遇到了一个小问题 - 我的mapper没有对输出进行排序&＃xff08;因为按常理来说不需要&＃xff09;&＃xff0c;但是我的reducer希望数据是按键排序的。我可以等着看最后的数字是怎么出来的&＃xff0c;但由于流式传输只涉及stdin输出到标准输入&＃xff0c;我有点好奇这个任务在Hadoop之外运行的速度可以有多快&＃xff08;我没有真正去比较&＃xff0c;针对简单的单节点集群; 我只是好奇&＃xff09;。而且我还在学习R&＃xff0c;所以接下来我编写了一个脚本来按记录键对行进行排序&＃xff1a;

#! /usr/bin/env Rscript
conn <- file("stdin", open&＃61;"r")
all.lines <- readLines(conn)
write(sort(all.lines), stdout())
close(conn)
在这种情况下&＃xff0c;我能明确回忆起我为什么这么喜欢R语言(即你就能明白R的强大之处)&＃xff01;接下来&＃xff0c;我将使用mapper的“测试”版来处理单个文件&＃xff1a;

./map.test.R | ./map.output.sorter.R | ./reduce.R
并为单个NIPS文件获得如下输出&＃xff08;缩写&＃xff09;&＃xff1a;

ALABAMA 72699735.21
ALASKA 6883209.62
ARIZONA 287482321.1
ARKANSAS 21579003.98
CALIFORNIA 1811342276.77
...
VIRGIN ISLANDS 1021750
WASHINGTON 239648997.97
WEST VIRGINIA 9925894.94
WISCONSIN 72752945.87
WYOMING 6232557.56
使用R在Hadoop中进行流式传输

现在我们有了一个mapper和一个reducer&＃xff0c;我们可以在Hadoop中处理整个数据集。我将处理与我之前的Hadoop-Java-Pig那个帖子中相同的数据集&＃xff0c;即2012年8月23日至12月26日的NIPS数据。正如在那篇文章中所展示的&＃xff0c;我以伪分布模式运行Hadoop&＃xff0c;使用来自HDFS的数据。当然&＃xff0c;这里的区别在于我指定了使用流式处理&＃xff0c;并提供了我的mapper和Reducer R脚本。我从Hadoop主目录启动&＃xff1a;

bin / hadoop jar $ HADOOP\_PREFIX / contrib / streaming / hadoop-streaming-1.1.0.jar -input / user / hduser / fannie-mae-nips -output / user / hduser / fannie-mae-nips -r-output -mapper /home/hduser/RScripts/map.R -reducer /home/hduser/RScripts/reduce.R
那么&＃xff0c;我的努力取得了什么结果&＃xff1f;从HDFS复制出我的结果文件&＃xff1a;

bin / hadoop dfs -copyToLocal / user / hduser / fannie-mae-nips -r-output / part-00000 rResults.txt
产生以下输出&＃xff08;这里是缩写&＃xff09;&＃xff1a;

ALABAMA 3242681838.89999
ALASKA 841797447.200001
ARIZONA 9340767235.06001
ARKANSAS 1452136751.9
CALIFORNIA 89114642822.0799
...
VERMONT 553060435.67
VIRGIN ISLANDS 33604327.46
VIRGINIA 12706719836.48
WASHINGTON 13194248475.54
WEST VIRGINIA 486889587.57
WISCONSIN 8140391871.79
WYOMING 720905726.84
我仍然保存着使用Java和Pig在这个相同的数据集上的输出; 仔细阅读此输出将凸显出以下输出&＃xff08;请注意&＃xff0c;由于数字以不同的格式输出&＃xff0c;因此我没有区分这些文件&＃xff09;&＃xff1a;

ALABAMA 3.242681838899994E9
ALASKA 8.417974472000003E8
ARIZONA 9.340767235060005E9
ARKANSAS 1.452136751900001E9
CALIFORNIA 8.91146428220799E10
....
VERMONT 5.530604356700001E8
VIRGIN ISLANDS 3.360432746000001E7
VIRGINIA 1.2706719836479996E10
WASHINGTON 1.319424847554002E10
WEST VIRGINIA 4.868895875700002E8
WISCONSIN 8.140391871790002E9
WYOMING 7.209057268400007E8
因此&＃xff0c;我成功地使用R和Hadoop流处理复制了使用Java和Pig进行的示例。

关于Hadoop和R的最终评论

如果你完全熟悉R&＃xff0c;你就会明白R并不是一种你为了分割输出和数字求和而选择的语言; 该语言及其库包含丰富的功能。这篇文章的重点主要是过一遍R与Hadoop流处理的机械式细节(即使用R与流处理的固定步骤)。R真正发光的地方在于&＃xff0c;如果是一些“繁重的工作”&＃xff0c;R很容易就能将其分解为Mapper风格和Reducer风格的任务。例如&＃xff0c;如果您正针对庞大的数据集进行线性回归操作&＃xff0c;使用了大量的变量&＃xff0c;或者如果您正在对大型数据集执行Shapiro-Wilk测试&＃xff0c;则可以将作业分解为并行任务&＃xff0c;最后将它们与Reducer相结合&＃xff0c;这将成为Hadoop/R协同工作的一个很好的例子。有关R中的并行计算的更多信息&＃xff0c;请查阅 R in a Nutshell&＃xff0c;特别是他在本章最后的注明的“在哪里了解更多”部分。