java通过jri 数据分析_SparkR安装部署及数据分析实例

╰半夏微凉° 2022-11-09 12:51 143阅读 0赞

1. SparkR的安装配置

1.1.       R与Rstudio的安装

1.1.1.           R的安装

我们的工作环境都是在Ubuntu下操作的，所以只介绍Ubuntu下安装R的方法：

1)  在/etc/apt/sources.list添加源

deb http://mirror.bjtu.edu.cn/cran/bin/linux/ubuntu precise/，

然后更新源apt-get update；

2)  通过apt-get安装：

sudo apt-get install r-base

1.1.2.           Rstudio的安装

官网有详细介绍：

sudo apt-get install gdebi-core

sudo apt-get install libapparmor1  \# Required only for Ubuntu, not Debian

wget http://download2.rstudio.org/rstudio-server-0.97.551-amd64.deb

sudo gdebi rstudio-server-0.97.551-amd64.deb

1.2.       rJava安装

1.2.1.           rJava介绍

rJava是一个R语言和Java语言的通信接口，通过底层JNI实现调用，允许在R中直接调用Java的对象和方法。

rJava还提供了Java调用R的功能，是通过JRI(Java/R Interface)实现的。JRI现在已经被嵌入到rJava的包中，我们也可以单独试用这个功能。现在rJava包，已经成为很多基于Java开发R包的基础功能组件。

正是由于rJava是底层接口，并使用JNI作为接口调用，所以效率非常高。在JRI的方案中，JVM通过内存直接加载RVM，调用过程性能几乎无损耗，因此是非常高效连接通道，是R和Java通信的首选开发包。

1.2.2.           rJava安装

1)  配置rJava环境

执行R CMD javareconf

root@testnode4:/home/payton\# R CMD javareconf

2)  启动R并安装rJava

root@testnode4:/home/payton\# R

> install.packages("rJava")

1.3.       SparkR的安装

1.3.1.           SparkR的代码下载

1.3.2.           SparkR的代码编译

1)  解压SparkR-pkg-master.zip，然后cd  SparkR-pkg-master/

2)  编译的时候需要指明Hadoop版本和Spark版本

SPARK\_HADOOP\_VERSION=2.4.1 SPARK\_VERSION=1.2.0 ./install-dev.sh

至此，单机版的SparkR已经安装完成。

1.3.3.           分布式SparkR的部署配置

1)  编译成功后，会生成一个lib文件夹，进入lib文件夹，打包SparkR为SparkR.tar.gz，这个是分布式SparkR部署的关键。

2)  由打包好的SparkR.tar.gz在各集群节点上安装SparkR

R CMD INSTALL SparkR.tar.gz

至此分布式SparkR搭建完成。

2. SparkR的运行

2.1.       SparkR的运行机制

SparkR是AMPLab发布的一个R开发包，为Apache Spark提供了轻量的前端。SparkR提供了Spark中弹性分布式数据集(RDD)的API，用户可以在集群上通过R shell交互性的运行job。SparkR集合了Spark 和R的优势，下面的这3幅图很好的阐释了SparkR的运行机制。

![7072044a2ae91c50eddf37d26bb90807.png][]

![77543762a218a01c54090774ab07c827.png][]

![ddb2cb86bc1dab1c25588754b262f7c8.png][]

2.2.       用SparkR 进行数据分析

2.2.1.           SparkR基本操作

首先介绍下SparkR的基本操作：

第一步，加载SparkR包

library(SparkR)

第二步，初始化Spark context

,sparkEnvir=list(spark.executor.memory="1g",spark.cores.max="10"))

第三步，读入数据，spark的核心是Resilient Distributed Dataset (RDD)，RDDS可以从Hadoop的InputFormats来创建(例如，HDFS文件)或通过转化其它RDDS。例如直接从HDFS读取数据为RDD的示例如下：

lines

另外，也可以通过parallelize函数从向量或列表创建RDD，如：

rdd

到了这里，那么我们就可以运用RDD的动作(actions)和转换(transformations)来对RDD进行操作并产生新的RDD；也可以很容易地调用R开发包，只需要在集群上执行操作前用includePackage读取R开发包就可以了(例：includePackage(sc, Matrix))；当然还可以把RDD转换为R语言格式的数据形式来对它进行操作。

具体可参见如下两个链接：

那么下面我们就通过两个示例来看下 SparkR是如何运行的吧。

2.2.2.           SparkR使用举例

1)Example1：word count

\#加载SparkR包

library(SparkR)\#初始化 Spark context

lines

words

wordCount

counts

output

for (wordcount inoutput) \{

cat(wordcount\[\[1\]\], ":", wordcount\[\[2\]\], "\\n")

2)Example2：logistic regression

\#加载SparkR包

library(SparkR)\#初始化 Spark context

appName='sparkr\_logistic\_regression',

sparkEnvir=list(spark.executor.memory='1g',

spark.cores.max="10"))\#从hdfs上读取txt文件， 该RDD由spark集群的4个分区构成

input\_rdd

minSplits=4)\#解析每个RDD元素的文本(在每个分区上并行)

dataset\_rdd

part

\})\#我们需要把数据集dataset\_rdd分割为训练集(train)和测试集(test)两部分，这里\#ptest为测试集的样本比例，如取ptest=0.2，即取dataset\_rdd的20%样本数作为测试\#集，80%的样本数作为训练集

split\_dataset

data\_test\_rdd

part\_test

\})\#用剩下的样本数创建训练集RDD

data\_train\_rdd

part\_train

\})\#返回测试集RDD和训练集RDD的列表

list(data\_test\_rdd, data\_train\_rdd)

\}\#接下来我们需要转化数据集为R语言的矩阵形式，并增加一列数字为1的截距项，\#将输出项y标准化为0/1的形式

get\_matrix\_rdd

matrix\_rdd

m\[,ncol(m)\]

\})

matrix\_rdd

\}\#由于该训练集中y的值为1与0的样本数比值为7:3，所以我们需要平衡1和0的样本\#数，使它们的样本数一致

balance\_matrix\_rdd

balanced\_matrix\_rdd

index

part

\})

balanced\_matrix\_rdd

\}\#分割数据集为训练集和测试集

dataset

matrix\_test\_rdd

matrix\_train\_rdd

cache(matrix\_test\_rdd)

cache(matrix\_train\_rdd)\#初始化向量theta

theta

hypot

\}\#损失函数的梯度计算

gCost

train

gradient\_rdd

p\_gradient

list(list(1, p\_gradient))

\})

agg\_gradient\_rdd

collect(agg\_gradient\_rdd)\[\[1\]\]\[\[2\]\]

\}\#由梯度下降算法优化损失函数\#alpha ：学习速率\#steps ：迭代次数\#tol ：收敛精度

alpha

while(T) \{

cat("step:",step,"\\n")

p\_gradient

theta

gradient

test

y\_pred

result

\})

result

corrects= length(result\[result==F\])

wrongs= length(result\[result==T\])

cat("\\ncorrects:",corrects,"\\n")

cat("wrongs:",wrongs,"\\n")

cat("accuracy:",corrects/length(y\_pred),"\\n")

[7072044a2ae91c50eddf37d26bb90807.png]: /images/20221023/f94146e7450049f0972607eaf8d56f3c.png
[77543762a218a01c54090774ab07c827.png]: /images/20221023/97dd7848fb4941da92d5d9e2e7cf299e.png
[ddb2cb86bc1dab1c25588754b262f7c8.png]: /images/20221023/6203b688d5df4efc8e86ac69210bb41d.png