如何使用CDSW在CDH中分布式运行所有R代码
温馨提示：要看高清无码套图，请使用手机打开并单击图片放大查看。
本文将是Fayson最近写的R系列的最后一篇文章了，大家且看且珍惜吧。
无需额外花费过多的学习成本，sparklyr（https://spark.rstudio.com）可以让R用户很方便的利用Apache Spark的分布式计算能力。之前Fayson介绍了什么是sparklyr，大家知道R用户可以编写几乎相同的代码运行在Spark之上实现本地或者分布式计算。
spark_apply的架构 (来自 https://github.com/rstudio/sparklyr/pull/728)
从sparklyr0.6（https://blog.rstudio.com/2017/07/31/sparklyr-0-6/）开始，你就可以通过spark_apply()运行R代码在Spark集群之上。换句话说，你可以用R写UDF。这样可以让你用你最喜欢的R包来访问Spark里的数据，比如仅在R中实现的特定的统计分析方法，或者像NLP的高级分析，等等。
因为目前spark_apply()的实现需要在工作节点上也安装R环境，在这篇文章里，我们将介绍如何在CDH集群中运行spark_apply()。我们会介绍两种方法：1.使用Parcel。2）使用conda环境。
选项1：使用Parcel安装R环境
创建和分发R的Parcel
Parcel（https://www.cloudera.com/documentation/enterprise/latest/topics/cm_ig_parcels.html）是一种二进制的分发格式，Cloudera Manager可以使用Parcel来分发CDH，Spark2，Kafka和需要运行在集群上的服务。Parcel很像.deb或者.rpm包。它可以让你通过Cloudera Manager的界面很容易的在CDH集群上安装特定的服务。使用这种方式的前提是CDH集群是使用Parcel方式安装的。
你可以到https://dl.bintray.com/chezou/Parcels/下载预先创建好的Parcel包，它包含了https://github.com/conda/conda-recipes/blob/master/r-packages/r-essentials/meta.yaml里的一些基础组件。注意你不能将Bintray直接设置为远程的Parcel仓库地址，所以你需要提前将你的Parcel上传到HTTP服务器。
如果你想使用其他的R包，可以使用R的Parcel构建工具。你可以用Docker创建你自己的Parcels，通过修改Dockerfile。
https://github.com/chezou/cloudera-parcel
将这些Parcels放到HTTP服务器或者指定的S3 bucket。然后你就可以在Cloudera Manager中添加Parcel的仓库地址。更多细节请参考：https://www.cloudera.com/documentation/enterprise/latest/topics/cm_ig_parcels.html#cmug_topic_7_11_5
在Spark的工作节点上运行R代码
当分发完R的Parcel包以后，就可以在工作节点上运行R代码。
注意：因为存在环境变量配置的问题：https://github.com/rstudio/sparklyr/issues/915，所以目前只能使用sparklyr的upstreamversion。最新的sparklyr 0.6.1没有这个功能。
以下是一个分布式执行R代码的例子
https://github.com/chezou/sparklyr-distribute
devtools::install_github("rstudio/sparklyr") library(sparklyr) config <- spark_config() config[["spark.r.command"]]<- "/opt/cloudera/parcels/CONDAR/lib/conda-R/bin/Rscript" config$sparklyr.apply.env.R_HOME <- "/opt/cloudera/parcels/CONDAR/lib/conda-R/lib/R" config$sparklyr.apply.env.RHOME <- "/opt/cloudera/parcels/CONDAR/lib/conda-R" config$sparklyr.apply.env.R_SHARE_DIR<- "/opt/cloudera/parcels/CONDAR/lib/conda-R/lib/R/share" config$sparklyr.apply.env.R_INCLUDE_DIR<- "/opt/cloudera/parcels/CONDAR/lib/conda-R/lib/R/include" sc <- spark_connect(master = "yarn-client" ,config = config) sdf_len(sc,5, repartition= 1 ) %>% spark_apply( function (e) I(e))r ## # Source: table<sparklyr_tmp_1cc757d61b8>[?? x 1] ## # Database: spark_connection ## id ## <dbl> ## 1 1 ## 2 2 ## 3 3 ## 4 4 ## 5 5
如果想要在分布式函数中使用R的包，sparklyr将这些包打包放在了本地的.libPaths()，然后使用SparkContext.addFile()函数将这些包分发到工作节点。如果是在spark_apply()中使用这些包则依赖于本地的代码，当然也可以按照下一个章节要介绍的使用Conda来分发他们。
更多官方文档资料： https://spark.rstudio.com/articles/guides-distributed-r.html#distributing-packages
选项2：使用conda环境
使用R创建conda环境
http://blog.cloudera.com/blog/2017/04/use-your-favorite-python-library-on-pyspark-cluster-with-cloudera-data-science-workbench/说明过conda虚拟环境可以打包R环境。
创建一个R的conda环境，使用zip压缩。
$ conda create -p ~/r_env --copy -y -q r-essentials -c r
# [Option] If you need additional package you can install as follows:
# $ source activate r_env
# $ Rscript -e 'install.packages(c("awesome-package"), lib = /home/cdsw/r_env/lib/R/library, dependencies = TRUE, repos="https://cran.r-project.org")'
# $ source deactivate与使用Parcel的差异是环境变量的设置，需要将r_env.zip设置为环境变量。虽然这种方式很灵活，但是需要每次创建Spark连接时都分发zip文件。全部代码请参考： https://github.com/chezou/sparklyr-distribute/blob/master/dist_sparklyr_conda.r
config <- spark_config()
config[["spark.r.command"]] <- "./r_env.zip/r_env/bin/Rscript"
config[["spark.yarn.dist.archives"]] <- "r_env.zip"
config$sparklyr.apply.env.R_HOME <- "./r_env.zip/r_env/lib/R"
config$sparklyr.apply.env.RHOME <- "./r_env.zip/r_env"
config$sparklyr.apply.env.R_SHARE_DIR <- "./r_env.zip/r_env/lib/R/share"
config$sparklyr.apply.env.R_INCLUDE_DIR <- "./r_env.zip/r_env/lib/R/include"然后你就可以在Spark的工作节点上运行R代码。
复杂的例子:使用spacyr做文本分析
注意：本版本目前不支持在spark_apply()中使用本地代码的R包。
在这个例子中，我们使用spacyr package（https://github.com/kbenoit/spacyr），这个包R绑定了spaCy（https://spacy.io），一个新的Python的NLP库。我们提取named-entity（https://en.wikipedia.org/wiki/Named-entity_recognition），比如person, place, organization等等，来自于Jane Austen’sbooks（https://www.rdocumentation.org/packages/janeaustenr/versions/0.1.5/topics/austen_books）。所有代码请参考:
https://github.com/chezou/spacyr-sparklyr
为spacyr准备conda环境
因为spacyr需要Python运行环境，在运行下面例子之前你需要安装Anaconda的Parcel。
使用https://github.com/chezou/spacyr-sparklyr/blob/master/install_spacy.sh这个脚本安装conda环境
用spark_apply()抽取named entities
Spark DataFrame有text的column，我们可以用下面的UDF抽取named entities
entities <- austen_tbl %>%
select(text) %>%
spark_apply(
function(e)
lapply(e, function(k) {
spacyr::spacy_initialize(python_executable="/opt/cloudera/parcels/Anaconda/bin/python")
parsedtxt <- spacyr::spacy_parse(as.character(k), lemma = FALSE)
spacyr::entity_extract(parsedtxt)
names = c("doc_id", "sentence_id", "entity", "entity_type"),
packages = FALSE)
entities %>% head(10) %>% collect()
## # A tibble: 10 x 4
## doc_id sentence_id entity entity_type
## <chr> <int> <chr> <chr>
## 1 text1 1 Jane Austen PERSON
## 2 text1 1 Dashwood ORG
## 3 text1 1 Sussex GPE
## 4 text1 2 Norland Park GPE
## 5 text1 4 Henry Dashwood PERSON
## 6 text1 4 Norland GPE
## 7 text1 5 Gentleman NORP
## 8 text1 7 Henry Dashwood PERSON
## 9 text1 8 Henry Dashwood PERSON
## 10 text1 11 Norland GPE划分entity类型的数量
library(ggplot2)
p <- entities %>%
collect() %>%
ggplot(aes(x=factor(entity_type)))
p <- p + scale_y_log10()
p + geom_bar()
前十名看每本书的人
persons <- entities %>%
filter(entity_type == "PERSON") %>%
group_by(doc_id, entity) %>%
select(doc_id, entity) %>%
count() %>%
arrange(doc_id, desc(n))
persons %>%
filter(doc_id == "text1") %>%
head(10) %>%
collect()
## # A tibble: 10 x 3
## # Groups: doc_id, entity [10]
## doc_id entity n
## <chr> <chr> <dbl>
## 1 text1 Elinor 662
## 2 text1 Marianne 538
## 3 text1 Edward 235
## 4 text1 Jennings 233
## 5 text1 Willoughby 200
## 6 text1 Lucy 172