[原]海纳百川 有容乃大：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商业使用请联系做者。