在Hadoop安装的集群上使用R

kennyut 发表于 Dev

肯尼特

我是R用户。我对Linux命令，PuTTY或Hadoop / Hive知之甚少。所以，如果我错了，请纠正我。

我现在正在与团队合作。他们在集群上有一个正在运行的Ubuntu系统。我可以使用PuTTY来访问此Ubuntu系统并通过使用代码来访问数据文件：

 user$hadoop fs -ls /datafolder/

或使用蜂巢：

 user$hive
 hive>use datafolder;
 hive>show tables;

相反，与我合作的团队对R知之甚少，因此他们希望我担任R部。我已经在群集上安装了R，并且还在R中安装了rJava HRive和其他软件包。（我不确定我是否正确执行了此操作，但是R似乎运行正常）。

现在我可以做一些测试了。我可以在R @群集上运行以下代码：

user$R
>install.packages(c('Rcpp', 'RJSONIO', 'bitops', 'digest', 'functional', 'stringr', 'plyr', 'reshape2','caTools'))

>Sys.setenv(HADOOP_CMD="/opt/cloudera/bin/hadoop")
>Sys.setenv(HADOOP_HOME="/opt/cloudera/lib/hadoop")
>Sys.setenv(HADOOP_STREAMING="/opt/cloudera/lib/hadoop-mapreduce/hadoop-streaming.jar")

>library(rmr2)
>library(rhdfs)
>hdfs.init()

测试：

>ints = to.dfs(1:10)
>calc = mapreduce(input = ints, map = function(k,v) cbind(v, v/2, 2*v))
>test <- from.dfs(calc)
>test

我可以使用from.dfs成功加载“测试”。看来我可以将虚拟数据集保存到Hadoop，并可以成功地从Hadoop取回它（对吗？）

现在，我的问题是，如何让R导入我可以从中看到的那些数据集

user$hadoop fs -ls /datafolder/

要么

>hive use datafolder;

躺着

这是单词计数的示例，其加载结果返回到R：

Sys.setenv(HADOOP_CMD="/usr/bin/hadoop")
    Sys.setenv(HADOOP_STREAMING="/opt/cloudera/parcels/CDH-5.1.0-1.cdh5.1.0.p0.53/lib/hadoop-0.20-mapreduce/contrib/streaming/hadoop-streaming.jar")
    Sys.setenv(JAVA_HOME="/usr/java/jdk1.7.0_55-cloudera")
    Sys.setenv(HADOOP_COMMON_LIB_NATIVE_DIR="/opt/cloudera/parcels/CDH-5.1.0-1.cdh5.1.0.p0.53/lib/hadoop/lib/native")
    Sys.setenv(HADOOP_OPTS="-Djava.library.path=HADOOP_HOME/lib")
    library(rhdfs)
    hdfs.init()
    library(rmr2)

    ## space and word delimiter
    map <- function(k,lines) {
      words.list <- strsplit(lines, '\\s')
      words <- unlist(words.list)
      return( keyval(words, 1) )
    }
    reduce <- function(word, counts) {
      keyval(word, sum(counts))
    }
    wordcount <- function (input, output=NULL) {
      mapreduce(input=input, output=output, input.format="text", map=map, reduce=reduce)
    }

    ## variables
    hdfs.root <- '/user/node'
    hdfs.data <- file.path(hdfs.root, 'data')
    hdfs.out <- file.path(hdfs.root, 'out')

    ## run mapreduce job
    ##out <- wordcount(hdfs.data, hdfs.out)
    system.time(out <- wordcount(hdfs.data, hdfs.out))

    ## fetch results from HDFS
    results <- from.dfs(out)
    results.df <- as.data.frame(results, stringsAsFactors=F)
    colnames(results.df) <- c('word', 'count')

    ##head(results.df)
    ## sorted output TOP10
    head(results.df[order(-results.df$count),],10)