由于装的sparkr是1.4版本的，老版本的很多函数已经不再适用了。

在2台服务器的组成的集群中测试了一版数据，熟悉下这个api的基本操作。​

libpath <- .libPaths()

libpath <- c(libpath, "/home/r/spark/spark-1.4.0-bin-hadoop2.4/R/lib")

.libPaths(libpath)

rm(libpath)

library(SparkR)

#没有默认找到变量 需单独设置，也可以在sparkR.init()里设置

Sys.setenv(SPARK_HOME="/home/r/spark/spark-1.4.0-bin-hadoop2.4/")

​#单主机启动

sc <- sparkR.init()

#集群启动

sc <- sparkR.init("spark://master机器ip:监听端口")

#失效​

# sc <- sparkR.init(sparkPackages="com.databricks:spark-csv_2.11:1.0.3")

sqlContext <- sparkRSQL.init(sc)

df <- createDataFrame(sqlContext, iris)

t <- head(summarize(groupBy(df, df$Species), count=n(df$Sepal_Length), m=sum(df$Sepal_Length)))

#读取数据将其转为json格式 以便sparkr读取​

pay.data <- read.table("/tmp/pay.dat", stringsAsFactors = F, sep="\t")

names(pay.data) <- c("user", "money")

df.pay <- createDataFrame(sqlContext, pay.data)

#createdataframe函数在原始data.frame很大时基本被huang住，不知道是什么原因

json <- apply(pay.data, 1, toJSON)

#   json <- toJSON(pay.data[i, ])

write.table(json, file="/tmp/1.json", col.names = F,

row.names=F, append=T, quote = F)

# }

write.table(josn1, file="http://183.60.122.213:7070/tmp/1.json", col.names = F, row.names=F, quote = F)

pay1.json <- read.df(sqlContext, "/home/r/spark/spark-1.4.0-bin-hadoop2.4/examples/src/main/resources/people.json",

"json")

#默认只支持默认只支持json和Parquet 格式文件，文件需要在work服务器上

pay.json <- read.df(sqlContext, "/tmp/1.json",

"json")

pay.json$money1 <- cast(pay.json$money, "double")

#将数据按帐号汇总统计后排序 输出​

pay.account <- agg(groupBy(pay.json, pay.json$user), money=sum(pay.json$money1),

num=n(pay.json$user))

pay.account1 <- arrange(pay.account, desc(pay.account$money), desc(pay.account$num))

write.df(pay.json, "/tmp/account1", "json")

分组统计了500w+的充值数据 并且排序后写成json文件到磁盘  时间是22s+​，比ddply要快，4700w耗时约26s,再大的数据暂时没有统计了。

理解的很粗浅，sparkr适用于r无法统计的大批数据的预处理，可以将简单预处理的汇总数据返回给R加以建模分析。其他还有待后续深入了解。

​

参考：

1）​简单介绍http://people.apache.org/~pwendell/spark-releases/latest/sparkr.html#sparkr-dataframes

2）常用函数http://people.apache.org/~pwendell/spark-releases/latest/api/R/index.html

​