R + RStudio(Server) + Spark/Hive 处理数据集

忘是亡心i 2022-11-29 05:59 364阅读 0赞

### 文章目录 ###

*  1 前言
 *  2 R 安装
 *   *  2.1 Mac 环境下
     *  2.2 CentOS 环境下安装 R
 *  3 RStudio 安装
 *   *  3.1 安装 RStudio Server
     *  3.2 连接 Spark
     *   *  3.2.1 local
         *  3.2.2 Cluster
 *  4 R + Spark处理数据集
 *   *  4.1 WordCount
 *  5 R + Hive 处理数据集
 *   *  5.1 连接 Hive
     *  5.2 实例：获取每个店铺排名前3的顾客记录

--------------------

# 1 前言 #

R是比较好的数据分析的语言工具，通过 R 的 DBI 又可以以 SQL 方式对数据进行处理。

R 是运行环境，RStudio 是开发工具。RStudio是为了方便使用R语言而诞生的，他能方便代码的编辑、调试以及图形的显示。所以需要先安装 R。

*  访问 R 语言官网：[https://www.r-project.org/][https_www.r-project.org]
 *  Linux/Mac/Windows系统版本的 R 下载：[https://cran.r-project.org/][https_cran.r-project.org]
 *  RStudio Download：[https://rstudio.com/products/rstudio/download/][https_rstudio.com_products_rstudio_download]
 *  RStudio Server Administration Guide：[https://docs.rstudio.com/ide/server-pro/][https_docs.rstudio.com_ide_server-pro]
 *  Sparklyr from RStrudio : [https://spark.rstudio.com/][https_spark.rstudio.com]

下面安装系统环境主要以 CentOS7 为主，但会提到 Mac 环境下的安装。后面安装的 RStudio 版本为目前最新的`1.3.1073-x86_64`，在这个版本中在 Spark 版本选择时有如下版本可供选择，为了少走弯路Spark 最好选择其中支持的安装部署，本次以 Spark 2.4.0 为例。  
![RStudio  支持的 Spark 版本][RStudio _ Spark]

# 2 R 安装 #

## 2.1 Mac 环境下 ##

# 1 下载，R-4.0.2 （为了和 RStudio 集成 要求 R 为 3.0.1 +）
    wget https://cran.r-project.org/bin/macosx/R-4.0.2.pkg
    
    # 2 安装。根据提示“下一步”安装即可，安装将占用 169.3M 的空间
    
    # 3 查看版本
    r --version

安装成功后启动台中有如下图标  
![R icon][]  
快速尝试一个小例子。可以双击上面的进入，也可以通过命令进入 console。

# 输入 r 进入 console
    Su-yuexi:~ yoreyuan$ R
    
    > age <- c(1,3,5,2,11,9,3,9,12,3)
    > weight <- c(4.4, 5.3, 7.2, 5.2, 8.5, 7.3, 6.0, 10.4, 10.2, 6.1)
    
    # 均值
    > mean(weight)
    [1] 7.06
    
    # 标准偏差
    > sd(weight)
    [1] 2.077498
    
    # 斜率
    > cor(age, weight)
    [1] 0.9075655
    
    # 绘图
    > plot(age, weight)

![plot images][]

## 2.2 CentOS 环境下安装 R ##

# 安装 dnf
    yum install epel-release -y
    yum install dnf
    dnf --version
    
    # 2 安装 R。使用 dnf 方式安装
    sudo dnf install R
    
    # 3 查看版本
    R --version

安装常用工具

# https://mirrors.tuna.tsinghua.edu.cn/CRAN
    R -e "install.packages('devtools', lib='/usr/lib64/R/library', repos='https://cran.r-project.org')"
    # R -e "devtools::install('.')"

# 3 RStudio 安装 #

访问 Rstudio 官网：[https://rstudio.com/products/rstudio/download/][https_rstudio.com_products_rstudio_download]，选择免费开源版 RStudio 下载即可。在选择是有两个版本可供选择，第一个是RStudio Desktop 版，这个适合桌面操作系统本机开发使用，桌面版不支持以 cluster 方式连接 Spark；第二个是 RStudio Server 版本，这个适合于服务器版开发使用，以账号方式通过浏览器页面执行操作。

![RStudio price][]

## 3.1 安装 RStudio Server ##

这里以 RStudio Server 版本介绍，在CentOS 7 系统中部署，下载 Linux Redhat 版本。

# 1 下载安装 RStudio Server
    # Install for Red Hat / CentOS 6-7
    wget https://download2.rstudio.org/server/centos6/x86_64/rstudio-server-rhel-1.3.1073-x86_64.rpm
    sudo yum install rstudio-server-rhel-1.3.1073-x86_64.rpm
    
    # 2 配置文件
    [root@cdh2 ~]# cd /etc/rstudio/
    [root@cdh2 rstudio]# ls
    rserver.conf  rsession.conf  themes
    
    ## 2.1 配置 rserver.conf
    sudo vi /etc/rstudio/rserver.conf
    ### 在上面的两个配置文件中添加如下配置
    ### 指定 R ，可以通过 which R 查看 R 的路径
    rsession-which-r=/usr/bin/R
    ### 配置端口。默认是 8787
    www-port=38787
    www-address=0.0.0.0
    
    ## 2.2 配置 rsession.conf 。可以默认
    ##sudo vi /etc/rstudio/rsession.conf 
    ### 在上面的两个配置文件中添加如下配置
    session-timeout-minutes=30
    r-cran-repos=https://mirrors.tuna.tsinghua.edu.cn/CRAN
    
    # 3 重启
    systemctl status rstudio-server
    rstudio-server restart
    
    # 5 浏览器访问。 rstudio / rstudio
    http://bigdata01:38787
    # rstudio-server 用户 密码
    # root 用户无法访问，使用其它 Linux 系统的普通用户登录

浏览器登录成功后显示如下，用户名和密码使用CentOS中的用户，但不能是 root 用户。  
![RStudio UI][]  
设置为国内镜像。为了使后面下载更加快速，选择国内 China 的镜像仓库即可，在上面的浏览器中依次点击 **Tools** \-> **Global Options** \-> **Packages** \-> **Management** \-> **Change**  
![添加国内镜像代理地址][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2dpdGh1Yl8zOTU3NzI1Nw_size_16_color_FFFFFF_t_70_pic_center]

## 3.2 连接 Spark ##

连接 Spark 可以点击 右上角区域的 **Connections** \-> **New Connection**，选择 Spark，在选择 Spark 时 Master 有两种方式：local 和 Cluster，local 方式是连接 RStudio 所在的本地的 Spark，Cluster 方式可以连接 Spark 集群。

![spark connection][]

### 3.2.1 local ###

`Master:` 选择 local 方式连接 Spark；`DB Interface:`默认使用 dplyr ；`Spark version:`选择支持的 Spark 版本；`Hadoop version:` 默认 2.7 ，也可以选择 2.6。

如果有提示，选择 Yes 升级或安装，注意左侧的 Jobs 区域，如果有发生如下的错误，可以如下方式解决：

*  错误 1：`ERROR: configuration failed for package ‘curl’` 解决
    
        wget -c https://curl.haxx.se/download/curl-7.65.0.tar.gz
        tar  -zxf  curl-7.65.0.tar.gz
        cd curl-7.65.0
        ./configure
        make
        sudo make install
        yum install -y libcurl-devel
 *  错误2：`ERROR: configuration failed for package ‘xml2’` 解决
    
        yum install -y libxml2-devel

如果显示如下`Package 'sparklyr' successfully installed.`，则表示 sparklyr 环境安装成功，可以正常选择连接 Spark进行连接。  
![New Connection][]  
选择 Spark 后，如果有提示`Connecting to Spark requires installation of an updated version of the shiny package. Do you want to install shiny now?`，我们选择 **Yes**。

第一次连接可以选择 local 方式，Spark 版本如前言部分提到的，我们选择一个支持的版本，Hadoop 默认为 2.7，也支持 2.6，

![New Connection + local][New Connection _ local]  
如果提示如下，手动执行 `spark_install(version = "2.4.0", hadoop_version = "2.7")`，

> library(sparklyr)
    > library(dplyr)
    
    Attaching package: ‘dplyr’
    
    The following objects are masked from ‘package:stats’:
    
        filter, lag
    
    The following objects are masked from ‘package:base’:
    
        intersect, setdiff, setequal, union
    
    > sc <- spark_connect(master = "local", version = "2.4.0")
    Error in spark_install_find(version, hadoop_version, latest = FALSE, hint = TRUE) : 
      Spark version not installed. To install, use spark_install(version = "2.4.0")

最后连接成功后如下图所示，接下来我们可以使用 R 通过 Spark 来处理我们的数据集  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2dpdGh1Yl8zOTU3NzI1Nw_size_16_color_FFFFFF_t_70_pic_center 1]

### 3.2.2 Cluster ###

当我们连接 Spark 时，`Master:` 选择 Cluster 时可能会报如下的错误，尽管我们已经在当前登录 RStudio 的用户的环境变量下配置 `SPARK_HOME` 依然会有如下提示，此时我们可以通过如下方式在 RStudio 中以 Cluster 方式连接 Spark。

> Connecting with a Spark cluster requires that you are on a system able to communicate with the cluster in both directions, and requires that the SPARK\_HOME environment variable refers to a locally installed version of Spark that is configured to communicate with the cluster.

> Your system doesn’t currently have the SPARK\_HOME environment variable defined. Please contact your system administrator to ensure that the server is properly configured to connect with the cluster.

![Connect to Spark 警告][Connect to Spark]

我们直接还是选择 local 方式，

# local 方式连接已经加载了。可略过
    library(sparklyr)
    library(dplyr)
    
    # 手动设置 Spark home
    Sys.setenv(SPARK_HOME = "/opt/spark-2.3.3-bin-hadoop2.7")
    
    # 重新以 cluster 方式连接 spark
    sc <- spark_connect(master = "spark://bigdata01:7077")

连接成功后可以在 Spark Master Web UI 中看到正在运行的 Application  
![Spark Master Web UI][]  
跳到 4.1 WordCount，执行一个 WordCount，  
![cluster 方式执行一个 wordcount][cluster _ wordcount]  
浏览器访问 Spark Master Web UI，查看 Job 的执行信息如下图：  
![Job 执行信息][Job]

# 4 R + Spark处理数据集 #

## 4.1 WordCount ##

# 第二个参数表示数据集的名称
    # 第三个参数标识数据在系统下的路径
    lines <- spark_read_text(sc, 
    		"rwordcount", 
    		path="/home/bigdata/spark/spark-2.4.0-bin-hadoop2.7/README.md")
    
    # 使用 ft_tokenizer 分割单词
    words <- lines %>% 
    	ft_tokenizer(input_col = "line",
                   output_col = "word_list")
    
    # 查看
    head(words, 4)
    
    # 对词进行 explode 打散
    words <- words %>%
      mutate(word = explode(word_list)) %>%
      select(word)
    
    head(words, 4)
    
    # 将上面的结果缓存到 Spark 内存中
    words <- words %>%
    compute("words")
    
    # 引入 DBI，它是 R 的一个数据库接口，直接使用 SQL 方法查询我们的数据
    > library(DBI)
    
    # 查询数据
    > dbGetQuery(sc, "SELECT word FROM words LIMIT 10")
          word
    1        #
    2   apache
    3    spark
    4         
    5    spark
    6       is
    7        a
    8     fast
    9      and
    10 general
    
    # 获取 word 的 TOP 10
    > dbGetQuery(sc, "SELECT word,COUNT(word) AS num FROM words 
    + 	GROUP BY word ORDER BY num DESC  LIMIT 10")
        word num
    1         72
    2    the  25
    3     to  19
    4  spark  16
    5    for  15
    6    and  11
    7      a  10
    8     ## 9
    9    you   8
    10   can   7
    
    
    # 退出
    > q()

![result][]

# 5 R + Hive 处理数据集 #

## 5.1 连接 Hive ##

通过 RStudio 使用 R 可以查询 Hive 中已有的数据。

R 中需要安装下面几个工具包，DBI 已经安装的可以略过，可以在 RStudio 右下侧的 **Packages** 中搜索查看，如果没有安装即可。

install.packages("rJava")
    install.packages("DBI")
    install.packages("RJDBC")

使用下面三个包，或者在工具中勾选，如下图所示。

library("rJava")
    library("DBI")
    library("RJDBC")

![rjava+dbi+rjdbc][rjava_dbi_rjdbc]

# 设置项目 home 路径
    # 将 hive-jdbc-x.x.x-standalone.jar 包放到下面路径下（在 hive安装目录的 lib 下可以得到此包）
    > setwd("/home/bigdata/spark")
    
    > hiveDriver <- JDBC("org.apache.hive.jdbc.HiveDriver", "hive-jdbc-2.1.1-cdh6.3.2-standalone.jar")
    > hiveConn <- dbConnect(hiveDriver, "jdbc:hive2://cdh3:10000", "hive用户名", "密码")
    
    # 查看表
    > dbGetQuery(hiveConn, "SHOW TABLES")

## 5.2 实例：获取每个店铺排名前3的顾客记录 ##

需求：假设目前有多个店铺，顾客访问任何一个店铺的任何一个商品都会产生一条日志。现在需要统计每个店铺访问量排名前 3 的顾客记录信息。

# 建表，因为没有返回值，可以忽略 `Unable to retrieve JDBC result set` 信息
    dbExecute(hiveConn, "CREATE TABLE `visit_t03` ( uid string comment '访客的用户id', shop_name string comment '被访问的店铺名称' ) ROW FORMAT DELIMITED FIELDS TERMINATED BY ',' STORED AS TEXTFILE")
    
    # 查看建的表信息
    > dbGetQuery(hiveConn, "DESC visit_t03")
       col_name data_type          comment
    1       uid    string     访客的用户id
    2 shop_name    string 被访问的店铺名称
    
    # 插入测试数据。其中第二个字段值 a、b、c …… 表示不同的店铺名
    dbExecute(hiveConn, "INSERT INTO visit_t03 VALUES ('u01', 'a'),('u02', 'b'),('u01', 'b'),('u01', 'a'),('u03', 'c'), ('u04', 'b'),('u01', 'a'),('u02', 'c'),('u05', 'b'),('u04', 'b'), ('u06', 'c'),('u02', 'c'),('u01', 'b'),('u02', 'a'),('u02', 'a'), ('u03', 'a'),('u05', 'a'),('u05', 'a'),('u05', 'a')")
    
    # 为实现需求，可以先按照下面的思路一步一步实现
    # step1：查询每个店铺被每个用户访问次数。
    # 这个比较简单，就是对商店名及用户 id 分组统计
    > dbGetQuery(hiveConn, "SELECT shop_name,uid,COUNT(*) v_count FROM visit_t03 + GROUP BY shop_name,uid")
       shop_name uid v_count
    1          a u01       3
    2          a u02       2
    3          a u03       1
    4          a u05       3
    5          b u01       2
    6          b u02       1
    7          b u04       2
    8          b u05       1
    9          c u02       2
    10         c u03       1
    11         c u06       1
    
    # step2：计算每个店铺被用户访问次数排名
    # 通过 step1 我们已经得到了每个店每个顾客的访问量信息了，这一步我们再对上一步的结果的访问进行排名
    # 这里的排名是针对每个店内的每个顾客排名，这里肯定需要用到开窗函数，
    # 那么会有人问这里可以使用 row_number() 吗?
    # 不适合，因为这里是取 top 前 3，需要排名，如果使用 row_number()，当访问量相同时会分为不同的排名，
    # 最好访问量相同排名也相同，所以这里使用 rank() ，大家也可以替换为 row_number() 执行下看下结果。
    > dbGetQuery(hiveConn, "SELECT shop_name,uid,v_count,rank() OVER(PARTITION BY shop_name ORDER BY v_count ) rk FROM ( + SELECT shop_name,uid,COUNT(*) v_count FROM visit_t03 + GROUP BY shop_name,uid + ) A")
       shop_name uid v_count rk
    1          a u03       1  1
    2          a u02       2  2
    3          a u05       3  3
    4          a u01       3  3
    5          b u05       1  1
    6          b u02       1  1
    7          b u04       2  3
    8          b u01       2  3
    9          c u06       1  1
    10         c u03       1  1
    11         c u02       2  3
    
    # step3: 取每个店铺排名前3的顾客数据
    # 这一步的目的及很简单了，直接取 step2 步结果中排名前 3 （rk <=3）的数据即可
    > dbGetQuery(hiveConn, "SELECT B.shop_name,B.uid,B.v_count FROM ( + SELECT shop_name,uid,v_count,rank() OVER(PARTITION BY shop_name ORDER BY v_count ) rk FROM ( + SELECT shop_name,uid,COUNT(*) v_count FROM visit_t03 + GROUP BY shop_name,uid + ) A + ) B WHERE rk <=3")
       b.shop_name b.uid b.v_count
    1            a   u03         1
    2            a   u02         2
    3            a   u01         3
    4            a   u05         3
    5            b   u02         1
    6            b   u05         1
    7            b   u01         2
    8            b   u04         2
    9            c   u03         1
    10           c   u06         1
    11           c   u02         2

中间执行的过程和通过 Hive 其它客户端执行是一样的，执行的记录也是 yarn 的 web ui （`http:/${RM_HOST}:8088/cluster`）上看到执行应用，如果中间需要停止某次提交，可以在这里查看对应的 application ID（或者 `yarn application -list`），通过 `yarn application -kill <Application ID>` 取消提交到 yarn 上的任务。

![yarn application list][]

[https_www.r-project.org]: https://www.r-project.org/
[https_cran.r-project.org]: https://cran.r-project.org/
[https_rstudio.com_products_rstudio_download]: https://rstudio.com/products/rstudio/download/
[https_docs.rstudio.com_ide_server-pro]: https://docs.rstudio.com/ide/server-pro/
[https_spark.rstudio.com]: https://spark.rstudio.com/
[RStudio _ Spark]: /images/20221124/20ead9ad98034483afa4de4805e6cb99.png
[R icon]: /images/20221124/c1420e2cfff145ec8678612d88415c33.png
[plot images]: /images/20221124/03a599478ed8479e9ae66d52cc0f7382.png
[RStudio price]: /images/20221124/461e1256bf2d4e2aaf04bf649d3d3b91.png
[RStudio UI]: /images/20221124/a37df92f42b74c3d95857e1445adeac3.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2dpdGh1Yl8zOTU3NzI1Nw_size_16_color_FFFFFF_t_70_pic_center]: /images/20221124/40719c04d23d4b99a7da4a2507fa086d.png
[spark connection]: /images/20221124/2f73f5b6602446ae9f7ab451f52969ab.png
[New Connection]: /images/20221124/886c66e0fd5e4cfd81bc63ac2da602d6.png
[New Connection _ local]: /images/20221124/20ebd9c8656c4a8aa775d195afc74391.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2dpdGh1Yl8zOTU3NzI1Nw_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20221124/c99d78a61688488aa382a50c372e3774.png
[Connect to Spark]: /images/20221124/6e7d9443548540f4b12a5fc94d6378d1.png
[Spark Master Web UI]: /images/20221124/405c4e6d1e7c40bd94d2f0953b0b0817.png
[cluster _ wordcount]: /images/20221124/530574d647cf4806a0e3fe1b6edccc66.png
[Job]: /images/20221124/edbc9ca8f3ef446ab83339b5ec2b7786.png
[result]: /images/20221124/845d6e4ebb3c46f8b025e578a08f3ea6.png
[rjava_dbi_rjdbc]: /images/20221124/31ef8138ae43498dafa55fb6f7a1a7a9.png
[yarn application list]: /images/20221124/9eaf4819050144d9831e427daea6582f.png