onetapbeyond / opencpu-spark-executor   1.0

#Apache Spark OpenCPU Executor (ROSE)

ROSE is an Apache Spark package offering access to the full scientific computing power of the R programming language to Spark batch and streaming applications on the JVM. This library is built on top of the opencpu-r-executor library, a lightweight solution for integrating R analytics executed on the OpenCPU server into any application running on the JVM.

Where Apache SparkR lets data scientists use Spark from R, ROSE is designed to let Scala and Java developers use R from Spark.

The popular Apache SparkR package provides a lightweight front-end for data scientists to use Apache Spark from R. This approach is ideally suited to investigative analytics, such as ad-hoc and exploratory analysis at scale.

The ROSE library attempts to provide the same R analytics capabilities available to Apache SparkR applications within traditional Spark applications on the JVM. It does this by exposing new analyze operations that execute R analytics on compatible RDDs. This new facility is designed primarily for operational analytics and can be used alongside Spark core, SQL, Streaming, MLib and GraphX.

If you need to query R machine-learning models, score R prediction models or leverage any other aspect of the R library within your Spark applications on the JVM then the ROSE library may be for you.

A number of example applications are provided to demonstrate the use of the ROSE library to deliver R analytics capabilities within any Spark solution.

• Hello, World! [ Scala ][ Java ]
• Batch Predictive Scoring Engine [ Scala ][ Java ]
• Streaming Predictive Scoring Engine [ Scala ][ Java ]
• Twitter Sentiment Analysis [ Scala ][ Java ]

libraryDependencies += "io.onetapbeyond" %% "opencpu-spark-executor_2.10" % "1.0"

compile 'io.onetapbeyond:opencpu-spark-executor_2.10:1.0'

Include the ROSE package in your Spark application using spark-shell, or spark-submit. For example:

$SPARK_HOME/bin/spark-shell --packages io.onetapbeyond:opencpu-spark-executor_2.10:1.0

This library exposes new analyze transformations on Spark RDDs of type RDD[OCPUTask]. The following sections demonstrate how to use these new RDD operations to execute R analytics directly within Spark batch and streaming applications on the JVM.

See the documentation on the underlying opencpu-r-executor library for details on building OCPUTask and handling OCPUResult.

For this example we assume an input dataRDD, then transform it to generate an RDD of type RDD[OCPUTask]. In this example each OCPUTask represents a fraud score prediction to be generated by the R function fraud::score when the RDD is eventually evaluated.

import io.onetapbeyond.opencpu.spark.executor.R._
import io.onetapbeyond.opencpu.r.executor._

val rTaskRDD = dataRDD.map(data => {
	OCPU.R()
		.pkg("fraud")
		.function("score")
		.input(data.asInput())
		.library()
	})

The set of OCPUTask within rTaskRDD can be scheduled for processing by calling the new analyze operation provided by ROSE on the RDD:

val rResultRDD = rTaskRDD.analyze

When rTaskRDD.analyze is evaluated by Spark the resultant rResultRDD is of type RDD[OCPUResult]. The fraud prediction score for the original OCPUTask are available within these OCPUResult. These values can be optionally cached, further processed or persisted per the needs of your Spark application.

Note, the use here of the R function fraud::score is simply representative of any R function or script available within the full set of R packages available on CRAN R, Bioconductor or on github.

For this example we assume an input stream dataStream, then transform it to generate a new stream with underlying RDDs of type RDD[OCPUTask]. In this example each OCPUTask represents a fraud score prediction to be generated by the R function fraud::score when the stream is eventually evaluated.

import io.onetapbeyond.opencpu.spark.executor.R._
import io.onetapbeyond.opencpu.r.executor._

val rTaskStream = dataStream.transform(rdd => {
	rdd.map(data => {
		OCPU.R()
			.pkg("fraud")
			.function("score")
			.input(data.asInput())
			.library()
		})	
	})

The set of OCPUTask within rTaskStream can be scheduled for processing by calling the new analyze operation provided by ROSE on each RDD within the stream:

val rResultStream = rTaskStream.transform(rdd => rdd.analyze)

When rTaskStream.transform is evaluated by Spark the resultant rResultStream has underlying RDDs of type RDD[OCPUResult]. The fraud prediction score for the original OCPUTask are available within these OCPUResult. These values can be optionally cached, further processed or persisted per the needs of your Spark application.

Note, the use here of the R function fraud::score is simply representative of any R function or script available within the full set of R packages available on CRAN R, Bioconductor or on github.

To understand how ROSE delivers the full scientific computing power of the R programming language to Spark applications on the JVM the following sections compare and constrast the deployment of traditional Scala, Java, Python and SparkR applications with Spark applications powered by the ROSE library.

The principal deployment requirement when working with ROSE is that your Spark cluster have access to one or more OpenCPU servers. Deployment options for those servers in the context of Spark are discussed in Application Deployment sections 3. and 4. that follow below.

####1. Traditional Scala | Java | Python Spark Application Deployment

Without ROSE library support, neither data scientists nor application developers have access to R's analytic capabilities within these types of application deployments.

####2. Traditional SparkR Application Deployment

While data scientists can leverage the computing power of Spark within R applications in these types of application deployments, these same R capabilities are not available to Scala, Java or Python developers.

Note, when working with Apache SparkR, the R runtime environment must be installed locally on each worker node on your cluster.

####3. Scala | Java + R (ROSE) Spark Application Deployment

Both data scientists and application developers working in either Scala or Java can leverage the full power of R using the ROSE library within these types of application deployments.

Using this deployment configuration, each worker node on the Spark cluster has its own dedicated OpenCPU server installed locally on the node. This configuration delivers optimal runtime throughput on the cluster. Note, this deployment configuration mirrors the configuration required by Apache SparkR, where the R runtime environment must be installed locally on each worker node on the cluster.

####4. Scala | Java + R (ROSE) Spark Application Deployment (Alternative)

Both data scientists and application developers working in either Scala or Java can leverage the full power of R using the ROSE library within these types of application deployments.

Using this deployment configuration, a cluster of one or more OpenCPU servers is maintained external to the Spark cluster. While maintaining these servers external to the Spark cluster potentially introduces runtime costs associated with networking latencies this approach does allow fast prototyping for ROSE-enabled Spark applications as no configuration changes need to be made to an existing Spark cluster to get up and running.

To take advantage of an external OpenCPU server cluster when working with ROSE simply pass an Array[String] of one or more server endpoints to the analyze transformation. For example, identify an external cluster of 3 OpenCPU servers for use by your Spark application:

val ocpuCluster = sc.broadcast(Array("http://1.1.x.x/ocpu",
									 "http://1.1.y.y/ocpu",
									 "http://1.1.z.z/ocpu"))
val rResultRDD = rTaskRDD.analyze(ocpuCluster.value)

As shown, the array of server endpoints should be maintained as a broadcast variable within your Spark application. Each OCPUTask within your ROSE application will automatically be distributed for execution at random across the OpenCPU server cluster.

See the LICENSE file for license rights and limitations (Apache License 2.0).