SparkR架构详解第一章：spark-submit提交R脚本

本帖最后由 regan 于 2016-10-10 17:13 编辑

1.sparkR中R脚本怎么提交的？2.sparkR中的R脚本提交到哪里？
3.SparkSubmit如何处理R脚本？
4.sparkR中是如何解析R脚本的？
5.SparkR整体架构？

R是数据科学家中最流行的编程语言和环境之一，在Spark中加入对R的支持是社区中较受关注的话题。作为增强Spark对数据科学家群体吸引力的最新举措，在Spark 1.4版本之后增加了R API（SparkR）。SparkR使得熟悉R的用户可以在Spark的分布式计算平台基础上结合R本身强大的统计分析功能和丰富的第三方扩展包，对大规模数据集进行分析和处理。
   1.R脚本怎么提交？
   使用spark-submit可以提交R脚本，格式为./spark-submit xxx.R
   2.R脚本提交到哪里？
   查看spark-submit脚本，有如下的语句：exec "${SPARK_HOME}"/bin/spark-class org.apache.spark.deploy.SparkSubmit "$@" ，这里有exec执行语句，执行的是org.apache.spark.deploy.SparkSubmit这个Scala对象中的main方法，注意这里的“$@”意思是将所有的参数以"arg1","arg2","arg3"。。。的形式传递给SparkSubmit的main方法
   3.SparkSubmit如何处理提交的R脚本？
   SparkSubmit对象的主要任务时根据提交的参数，判断程序的运行模式及判断是提交的是phyton脚本，还是R脚本，或是Jar包。这里怎么判断是提交的R脚本呢？有一个isR方法：

private[deploy] def isR(res: String): Boolean = { res != null && res.endsWith(".R") || res == SPARKR_SHELL
}

看到了吧，这里直接判断后缀，如果后缀为.R那么提交的是一个R脚本，判断是R脚本后，怎样处理呢？JVM肯定解析不了R语法，要解析R语法还得使用RScript才行，因此这里需要调用RScript进程，运行R脚本。怎么调的呢？
   4.SparkSubmit中怎么调用RScript进程解析R文件的？
   直接看源代码吧！
   if (args.isR && deployMode == CLIENT) {
  if (args.primaryResource == SPARKR_SHELL) {
args.mainClass = "org.apache.spark.api.r.RBackend"
  } else {
// If an R file is provided, add it to the child arguments and list of files to deploy.
// Usage: RRunner <main R file> [app arguments]
args.mainClass = "org.apache.spark.deploy.RRunner"
args.childArgs = ArrayBuffer(args.primaryResource) ++ args.childArgs
args.files = mergeFileLists(args.files, args.primaryResource)
  }
}
  这里会启动一个RRunner用于启动RScript进程，在RScript进程中会运行提交的R脚本。  RRunner怎样启动RScript进程的？  RRunner中会启动ProcessBuilder构建并运行Rscript  5.sparkR的整体构架是什么？

SparkR的架构类似于PySpark，Driver端除了一个JVM进程（包含一个SparkContext）外，还有起一个R的进程，这两个进程通过Socket进行通信，用户可以提交R语言代码，R的进程会执行这些R代码，当R代码调用Spark相关函数时，R进程会通过Socket触发JVM中的对应任务。
当R进程向JVM进程提交任务的时候，R会把子任务需要的环境(enclosing environment)进行打包，并发送到JVM的driver端。
通过R生成的RDD都会是RRDD类型，当触发RRDD的action时，Spark的执行器会开启一个R进程(worker.R)，执行器和R进程通过Socket进行通信。执行器会把任务和所需的环境发送给R进程，R进程会加载对应的package，执行任务，并返回结果。

下面是介绍一些执行SparkR的详细流程：
  1. SparkSubmit判断是SparkR类型的任务，启动RRunner
  2. RRunner首先启动RBackend，然后再启动R进程执行用户的R脚本
  3. RBackend开启Socket Server，等待R进程链接
  4. R进程通过Socket连接到RBackend
  5. R脚本生成SparkContext，通过Socket通信，会在JVM中生成JavaSparkContext
  6. R脚本触发RRDD的action，通过Socket通信，Driver启动executor，并执行对于的任务
  7. 在每个执行器中都会调用RRDD的compute函数，来计算RRDD中的数据
  8. compute函数首先开启Socket Server，等待R进程链接，然后启动一个R进程(worker.R)