如何使用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
##  7  text1       Dashwood   159
##  8  text1        Ferrars   104
##  9  text1 Lady Middleton    80
## 10  text1         Palmer    74

我应该使用哪个选项？

一般来说，建议选择选项1，因为你不需要每次分发R的环境，而且构建包含所有包的Parcel节约了很多时间，而不用纠结于某一个包。目前，RStudio有OS（https://spark.rstudio.com/articles/guides-distributed-r.html#requirements）的限制，但你可以设置packages= FALSE来绕过因为OS不同的问题。

另外，选项2可以灵活的后面再安装包，但用起来会麻烦一些。如果只是仅仅使用R包，这个方法也行，但当你还想使用rJava等源生的扩展包的时候，则比较难准备好环境。

总结

本文主要是介绍了如何使用sparklyr在Spark工作节点上运行和分发R代码。因为spark_apply()方法需要在工作节点上安装R，我们介绍了两种方法可以让你在CDH集群和CDSW上运行spark_apply()。你可以根据你想要的进行选择。如果需要稳定，可以选择选项1：Parcel的方法。如果需要灵活，则可以选择选项2：conda环境。

不仅只是执行dplyr，同时你可以分发你本地的R代码到Spark集群。这样可以让你将你的R技能充分应用到分布式计算框架上。

参考

https://blog.cloudera.com/blog/2017/09/how-to-distribute-your-r-code-with-sparklyr-and-cdsw/

醉酒鞭名马，少年多浮夸！ 岭南浣溪沙，呕吐酒肆下！挚友不肯放，数据玩的花！ 温馨提示：要看高清无码套图，请使用手机打开并单击图片放大查看。

推荐关注Hadoop实操，第一时间，分享更多Hadoop干货，欢迎转发和分享。

原创文章，欢迎转载，转载请注明：转载自微信公众号Hadoop实操