[原]海納百川 有容乃大：SparkR與Docker的機器學習實戰

題圖爲美國尼米茲核動力航空母艦

介紹

大數據時代，咱們經常面對海量數據而頭疼。做爲學統計出身的人，咱們想折騰大數據但又不想學習Hadoop或者Java，咱們更傾向於把精力放在建模和算法設計上，SparkR和Docker的完美結合，讓R的計算直接從一架戰鬥機的當兵做戰華麗轉變爲一個航空母艦戰鬥羣！不只僅簡化了分佈式計算的操做，還簡化了安裝部署的環節，咱們只幾乎不須要作什麼改動就能夠直接運用R中的data frame進行分佈式的計算。

什麼是SparkR

參考前文 打造大數據產品：Shiny的Spark之旅,咱們能夠知道，SparkR是一個爲R提供了輕量級的Spark前端的R包。 SparkR提供了一個分佈式的data frame數據結構，解決了 R中的data frame只能在單機中使用的瓶頸，它和R中的data frame 同樣支持許多操做，好比select,filter,aggregate等等。（相似dplyr包中的功能）這很好的解決了R的大數據級瓶頸問題。 SparkR也支持分佈式的機器學習算法，好比使用MLib機器學習庫。

什麼是Docker

參考前文 打造數據產品的快速原型：Shiny的Docker之旅，咱們也能夠知道，Docker是一種相似於虛擬機的技術，主要解決標準化快速部署的問題，在Docker中安裝的軟件和主機中的軟件能夠徹底隔離，並經過Daocloud或者hub.docker.com等雲服務快速創建Docker倉庫，快速複用Docker鏡像。Docker已經不只僅是DevOps人員手中的神器了，每個開發者都應該學會如何使用Docker。

爲何要結合SparkR和Docker

SparkR的精髓在於分佈式計算，而Docker的精髓在於標準容器的拓展性，SparkR和Docker的組合充分結合了兩者各自的優勢，將分佈式應用底層化繁爲簡，爲高層計算直接暴露接口，給科學計算節省了大量時間。

部署

本文將經過Docker講解如何快速部署SparkR－RStudio容器，並經過一些簡單的機器學習例子展現如何使用這個航母級別的組合拳。

步驟一：安裝Docker和Daocloud

因爲國內的鏡像質量不夠高，國外的鏡像下載速度比較慢，出於試驗的考慮，建議你們能夠嘗試使用Daocloud的鏡像加速服務。

首先，咱們須要在Daocloud註冊一個帳號，而後選擇鏡像加速，根據指示選擇主機並安裝Docker和Daocloud加速器。

步驟二：安裝Spark－RStudio

感謝 vinicius85 在GitHub上的開源貢獻，爲咱們已經作好了 Spark1.6+R＋RStduio的鏡像，咱們利用daocloud加速拉取鏡像。

dao pull vinicius85/spark-rstudio

以daemon形式運行容器，暴露Rstudio－server默認的8787端口, 並持久化docker內的/srv目錄下的全部文件做爲通信。

docker run -d -v /home/docker:/srv -p 8787:8787 --name sparkrstudio vinicius85/sparkr-rstudio

或者經過下面最新的方式安裝

docker run -d -p 8787:8787 --name financer index.tenxcloud.com/7harryprince/sparkr-rstudio​

步驟三：配置RStudio登錄帳號

參考前文 R語言工程化實踐：RStudio Server環境快速配置教程

docker exec -d sparkrstudio bash命令表示以daemon形式執行容器中的shell腳本

咱們設置一下RStudio－Server的帳號密碼

docker exec -d sparkrstudio bash adduser harryzhu # 設置新用戶名
docker exec -d sparkrstudio bash passwd harryzhu # 設置該用戶的密碼

步驟四：登錄RStudio

ifconfig命令能夠查看到Docker當前的IP地址，透過這個IP，咱們能夠訪問到RStudio－Server。

好比：

查看資源佔用狀況

docker stats sparkrstudio

CONTAINER           CPU %               MEM USAGE / LIMIT     MEM %               NET I/O               BLOCK I/O
sparkrstudio        4.50%               481.3 MB / 5.039 GB   9.55%               133.6 kB / 117.4 kB   3.252 MB / 135.2 kB

機器學習示例：

出於演示的考慮，這裏引用並稍微改進了 tcosta 完成的一個邏輯迴歸的例子：

初始化

使用SparkR以前，咱們須要肯定，咱們的容器內存要在2G以上，若是用AWS的乞丐版套裝，立刻就會報內存不足的錯誤。

Error in sparkR.init(master = "local") :
   JVM is not ready after 10 seconds

若是內存不足，能夠退出docker而且在虛擬機中從新提升docker的內存和cpu的配置。

# 配置環境變量
Sys.setenv(SPARK_HOME="/opt/spark-1.6.0-bin-hadoop2.6")
 
.libPaths(c(file.path(Sys.getenv("SPARK_HOME"), "R", "lib"), .libPaths()))
 
Sys.setenv(JAVA_HOME="/usr/lib/jvm/java-8-oracle/")

# 加載 SparkR包
library(SparkR)
 
# 初始化RRD
#sc <- sparkR.init(master = "local")
#sqlContext <- sparkRSQL.init(sc)
# spark 2.0 後改成
sc <- sparkR.session(master = "local")

# 建立DataFrame
#mtcarsDF <- createDataFrame(sqlContext, mtcars)
mtcarsDF <- createDataFrame( mtcars)
head(mtcarsDF)

mpg cyl disp  hp drat    wt  qsec vs am gear carb
1 21.0   6  160 110 3.90 2.620 16.46  0  1    4    4
2 21.0   6  160 110 3.90 2.875 17.02  0  1    4    4
3 22.8   4  108  93 3.85 2.320 18.61  1  1    4    1
4 21.4   6  258 110 3.08 3.215 19.44  1  0    3    1
5 18.7   8  360 175 3.15 3.440 17.02  0  0    3    2
6 18.1   6  225 105 2.76 3.460 20.22  1  0    3    1

邏輯迴歸

model <- glm(vs ~ mpg + disp + hp + wt , data = mtcarsDF, family = "binomial")# 邏輯迴歸
# model <- glm(vs ~ mpg + disp + hp + wt , data = mtcarsDF, family = "gaussian")# 線性迴歸
predictions <- predict(model, newData = mtcarsDF )
modelPrediction <- select(predictions, "vs", "prediction")
head(modelPrediction)

vs prediction
1  0 0.58006945
2  0 0.64060709
3  1 0.72468718
4  1 0.47803842
5  0 0.06070972
6  1 0.54994276

模型評估

# error變量: 觀測值和預測值的差值
modelPrediction$error <- abs(modelPrediction$vs - modelPrediction$prediction)
 
# modelPrediction 如今對 SQLContext 是可見的
# registerTempTable(modelPrediction, "modelPrediction")
# Spark 2.0 以後api改成
createTempTable(modelPrediction, "modelPrediction")
 
#num_errors <- sql(sqlContext, "SELECT count(error) FROM modelPrediction WHERE error = 1")
#total_errors <- sql(sqlContext, "SELECT count(error) FROM modelPrediction")

num_errors <- sql( "SELECT count(error) FROM modelPrediction WHERE error = 1")
total_errors <- sql( "SELECT count(error) FROM modelPrediction")

# 模型錯誤率
training_acc <- collect(num_errors) / collect(total_errors)
training_acc

_c0
1   0

參考資料

• 打造數據產品的快速原型：Shiny的Docker之旅

• R語言工程化實踐：RStudio Server環境快速配置教程

• 打造大數據產品：Shiny的Spark之旅

• Tiago Vinícius: SparkR 1.5 MLlib Logistic Regression Example

• SparkR: Distributed data frames with Spark and R

• SparkR R frontend for Spark

• 劉志強, et al. "基於 SparkR 的分類算法並行化研究." Journal of Frontiers of Computer Science and Technology 9.11 (2015): 1281-1294.

• 用Parallel和foreach包玩轉並行計算

• Zeppelin跟SparkR使用spark 1.5+的分析平臺建置

• Fast learn scala

• Spark在美團的實踐

做爲分享主義者(sharism)，本人全部互聯網發佈的圖文均聽從CC版權，轉載請保留做者信息並註明做者 Harry Zhu 的 FinanceR專欄:https://segmentfault.com/blog...，若是涉及源代碼請註明GitHub地址：https://github.com/harryprince。微信號: harryzhustudio商業使用請聯繫做者。