1. SparkR的安裝配置
1.1. R與Rstudio的安裝
1.1.1. R的安裝
我們的工作環境都是在Ubuntu下操作的，所以只介紹Ubuntu下安裝R的方法：
1） 在/etc/apt/sources.list添加源
deb http://mirror.bjtu.edu.cn/cran/bin/linux/ubuntu precise/，
然后更新源apt-get update；
2） 通過apt-get安裝：
sudo apt-get install r-base
1.1.2. Rstudio的安裝
官網有詳細介紹：
http://www.rstudio.com/products/rstudio/download-server/
sudo apt-get install gdebi-core
sudo apt-get install libapparmor1 # Required only for Ubuntu, not Debian
wget http://download2.rstudio.org/rstudio-server-0.97.551-amd64.deb
sudo gdebi rstudio-server-0.97.551-amd64.deb
1.2. rJava安裝
1.2.1. rJava介紹
rJava是一個R語言和Java語言的通信接口，通過底層JNI實現調用，允許在R中直接調用Java的對象和方法。
rJava還提供了Java調用R的功能，是通過JRI(Java/R Interface)實現的。JRI現在已經被嵌入到rJava的包中，我們也可以單獨試用這個功能。現在rJava包，已經成為很多基於Java開發R包的基礎功能組件。
正是由於rJava是底層接口，並使用JNI作為接口調用，所以效率非常高。在JRI的方案中，JVM通過內存直接加載RVM，調用過程性能幾乎無損耗，因此是非常高效連接通道，是R和Java通信的首選開發包。
1.2.2. rJava安裝
1） 配置rJava環境
執行R CMD javareconf
root@testnode4:/home/payton# R CMD javareconf
2） 啟動R並安裝rJava
root@testnode4:/home/payton# R
> install.packages("rJava")
1.3. SparkR的安裝
1.3.1. SparkR的代碼下載
從網頁下載代碼SparkR-pkg-master.zip https://github.com/amplab-extras/SparkR-pkg
1.3.2. SparkR的代碼編譯
1） 解壓SparkR-pkg-master.zip，然后cd SparkR-pkg-master/
2） 編譯的時候需要指明Hadoop版本和Spark版本
SPARK_HADOOP_VERSION=2.4.1 SPARK_VERSION=1.2.0 ./install-dev.sh
至此，單機版的SparkR已經安裝完成。
1.3.3. 分布式SparkR的部署配置
1） 編譯成功后，會生成一個lib文件夾，進入lib文件夾，打包SparkR為SparkR.tar.gz，這個是分布式SparkR部署的關鍵。
2） 由打包好的SparkR.tar.gz在各集群節點上安裝SparkR
R CMD INSTALL SparkR.tar.gz
至此分布式SparkR搭建完成。
2. SparkR的運行
2.1. SparkR的運行機制
SparkR是AMPLab發布的一個R開發包，為Apache Spark提供了輕量的前端。SparkR提供了Spark中彈性分布式數據集（RDD）的API，用戶可以在集群上通過R shell交互性的運行job。SparkR集合了Spark 和R的優勢，下面的這3幅圖很好的闡釋了SparkR的運行機制。
2.2. 用SparkR 進行數據分析
2.2.1. SparkR基本操作
首先介紹下SparkR的基本操作：
第一步，加載SparkR包
library(SparkR)
第二步，初始化Spark context
sc <- sparkR.init(master=" spark://localhost:7077"
,sparkEnvir=list(spark.executor.memory="1g",spark.cores.max="10"))
第三步，讀入數據，spark的核心是Resilient Distributed Dataset (RDD)，RDDS可以從Hadoop的InputFormats來創建（例如，HDFS文件）或通過轉化其它RDDS。例如直接從HDFS讀取數據為RDD的示例如下：
lines <- textFile(sc, "hdfs://sparkR_test.txt")
另外，也可以通過parallelize函數從向量或列表創建RDD，如：
rdd <- parallelize(sc, 1:10, 2)
到了這里，那么我們就可以運用RDD的動作（actions）和轉換（transformations）來對RDD進行操作並產生新的RDD；也可以很容易地調用R開發包，只需要在集群上執行操作前用includePackage讀取R開發包就可以了（例：includePackage(sc, Matrix)）；當然還可以把RDD轉換為R語言格式的數據形式來對它進行操作。
具體可參見如下兩個鏈接：
http://amplab-extras.github.io/SparkR-pkg/
https://github.com/amplab-extras/SparkR-pkg/wiki/SparkR-Quick-Start
那么下面我們就通過兩個示例來看下 SparkR是如何運行的吧。
2.2.2. SparkR使用舉例
1） Example1：word count
# 加載SparkR包 library(SparkR) # 初始化 Spark context sc <- sparkR.init(master="spark://集群ip:7077" ,sparkEnvir=list(spark.executor.memory="1g",spark.cores.max="10")) # 從HDFS上讀取文件 lines <- textFile(sc, "hdfs://集群ip:8020/tmp/sparkR_test.txt") # 按分隔符拆分每一行為多個元素，這里返回一個序列 words<-flatMap(lines,function(line) {strsplit(line,"\\|")[[1]]}) # 使用 lapply 來定義對應每一個RDD元素的運算，這里返回一個（K，V)對 wordCount <-lapply(words, function(word) { list(word, 1L) }) # 對（K，V）對進行聚合計算 counts<-reduceByKey(wordCount,"+",2L) # 以數組的形式，返回數據集的所有元素 output <- collect(counts) # 按格式輸出結果 for (wordcount in output) { cat(wordcount[[1]], ": ", wordcount[[2]], "\n") }
2） Example2：logistic regression
# 加載SparkR包 library(SparkR) # 初始化 Spark context sc <- sparkR.init(master="集群ip:7077", appName='sparkr_logistic_regression', sparkEnvir=list(spark.executor.memory='1g', spark.cores.max="10")) # 從hdfs上讀取txt文件， 該RDD由spark集群的4個分區構成 input_rdd <- textFile(sc, "hdfs://集群ip:8020/user/payton/german.data-numeric.txt", minSplits=4) # 解析每個RDD元素的文本（在每個分區上並行） dataset_rdd <- lapplyPartition(input_rdd, function(part) { part <- lapply(part, function(x) unlist(strsplit(x, '\\s'))) part <- lapply(part, function(x) as.numeric(x[x != ''])) part }) # 我們需要把數據集dataset_rdd分割為訓練集（train）和測試集（test）兩部分，這里 # ptest為測試集的樣本比例，如取ptest=0.2，即取dataset_rdd的20%樣本數作為測試 # 集，80%的樣本數作為訓練集 split_dataset <- function(rdd, ptest) { #以輸入樣本數ptest比例創建測試集RDD data_test_rdd <- lapplyPartition(rdd, function(part) { part_test <- part[1:(length(part)*ptest)] part_test }) # 用剩下的樣本數創建訓練集RDD data_train_rdd <- lapplyPartition(rdd, function(part) { part_train <- part[((length(part)*ptest)+1):length(part)] part_train }) # 返回測試集RDD和訓練集RDD的列表 list(data_test_rdd, data_train_rdd) } # 接下來我們需要轉化數據集為R語言的矩陣形式，並增加一列數字為1的截距項， # 將輸出項y標准化為0/1的形式 get_matrix_rdd <- function(rdd) { matrix_rdd <- lapplyPartition(rdd, function(part) { m <- matrix(data=unlist(part, F, F), ncol=25, byrow=T) m <- cbind(1, m) m[,ncol(m)] <- m[,ncol(m)]-1 m }) matrix_rdd } # 由於該訓練集中y的值為1與0的樣本數比值為7:3，所以我們需要平衡1和0的樣本 # 數，使它們的樣本數一致 balance_matrix_rdd <- function(matrix_rdd) { balanced_matrix_rdd <- lapplyPartition(matrix_rdd, function(part) { y <- part[,26] index <- sample(which(y==0),length(which(y==1))) index <- c(index, which(y==1)) part <- part[index,] part }) balanced_matrix_rdd } # 分割數據集為訓練集和測試集 dataset <- split_dataset(dataset_rdd, 0.2) # 創建測試集RDD matrix_test_rdd <- get_matrix_rdd(dataset[[1]]) # 創建訓練集RDD matrix_train_rdd <- balance_matrix_rdd(get_matrix_rdd(dataset[[2]])) # 將訓練集RDD和測試集RDD放入spark分布式集群內存中 cache(matrix_test_rdd) cache(matrix_train_rdd) # 初始化向量theta theta<- runif(n=25, min = -1, max = 1) # logistic函數 hypot <- function(z) { 1/(1+exp(-z)) } # 損失函數的梯度計算 gCost <- function(t,X,y) { 1/nrow(X)*(t(X)%*%(hypot(X%*%t)-y)) # 定義訓練函數 train <- function(theta, rdd) { # 計算梯度 gradient_rdd <- lapplyPartition(rdd, function(part) { X <- part[,1:25] y <- part[,26] p_gradient <- gCost(theta,X,y) list(list(1, p_gradient)) }) agg_gradient_rdd <- reduceByKey(gradient_rdd, '+', 1L) # 一次迭代聚合輸出 collect(agg_gradient_rdd)[[1]][[2]] } # 由梯度下降算法優化損失函數 # alpha ：學習速率 # steps ：迭代次數 # tol ：收斂精度 alpha <- 0.1 tol <- 1e-4 step <- 1 while(T) { cat("step: ",step,"\n") p_gradient <- train(theta, matrix_train_rdd) theta <- theta-alpha*p_gradient gradient <- train(theta, matrix_train_rdd) if(abs(norm(gradient,type="F")-norm(p_gradient,type="F"))<=tol) break step <- step+1 } # 用訓練好的模型預測測試集信貸評測結果（“good”或“bad”），並計算預測正確率 test <- lapplyPartition(matrix_test_rdd, function(part) { X <- part[,1:25] y <- part[,26] y_pred <- hypot(X%*%theta) result <- xor(as.vector(round(y_pred)),as.vector(y)) }) result<-unlist(collect(test)) corrects = length(result[result==F]) wrongs = length(result[result==T]) cat("\ncorrects: ",corrects,"\n") cat("wrongs: ",wrongs,"\n") cat("accuracy: ",corrects/length(y_pred),"\n")