r - 使用 SparkR JVM 从 Scala jar 文件中调用方法

Question

我希望能够将 DataFrames 打包在 Scala jar 文件中并在 R 中访问它们。最终目标是创建一种方法来访问 Python、R 和 Scala 中特定且经常使用的数据库表，而无需为每个表编写不同的库.

为此，我在 Scala 中制作了一个 jar 文件，其中包含使用 SparkSQL 库来查询数据库并获取我想要的 DataFrame 的函数。我希望能够在 R 中调用这些函数而无需创建另一个 JVM，因为 Spark 已经在 R 中的 JVM 上运行。但是，SparkR API 中没有公开 Spark 使用的 JVM。为了使其可访问并使 Java 方法可调用，我修改了 SparkR 包中的“backend.R”、“generics.R”、“DataFrame.R”和“NAMESPACE”并重建了该包:

在“backend.R”中，我制作了“callJMethod”和“createJObject”正式方法:

  setMethod("callJMethod", signature(objId="jobj", methodName="character"), function(objId, methodName, ...) {
  stopifnot(class(objId) == "jobj")
  if (!isValidJobj(objId)) {
    stop("Invalid jobj ", objId$id,
         ". If SparkR was restarted, Spark operations need to be re-executed.")
  }
  invokeJava(isStatic = FALSE, objId$id, methodName, ...)
})


  setMethod("newJObject", signature(className="character"), function(className, ...) {
  invokeJava(isStatic = TRUE, className, methodName = "<init>", ...)
})

我修改了“generics.R”以包含这些函数:

#' @rdname callJMethod
#' @export
setGeneric("callJMethod", function(objId, methodName, ...) { standardGeneric("callJMethod")})

#' @rdname newJobject
#' @export
setGeneric("newJObject", function(className, ...) {standardGeneric("newJObject")})

然后我将这些函数的导出添加到 NAMESPACE 文件中:

export("cacheTable",
   "clearCache",
   "createDataFrame",
   "createExternalTable",
   "dropTempTable",
   "jsonFile",
   "loadDF",
   "parquetFile",
   "read.df",
   "sql",
   "table",
   "tableNames",
   "tables",
   "uncacheTable",
   "callJMethod",
   "newJObject")

这让我可以在不启动新 JVM 的情况下调用我编写的 Scala 函数。

我编写的 scala 方法返回 DataFrames，返回时它们是 R 中的“jobj”，但 SparkR DataFrame 是一个环境 + 一个 jobj。为了将这些 jobj DataFrames 转换为 SparkR DataFrames，我使用了“DataFrame.R”中的 dataFrame() 函数，我也可以按照上述步骤访问该函数。

然后，我能够从 R 访问我在 Scala 中“构建”的 DataFrame，并在该 DataFrame 上使用 SparkR 的所有函数。我想知道是否有更好的方法来制作这样一个跨语言库，或者是否有任何理由不应该公开 Spark JVM？

Answer 1

any reason the Spark JVM should not be public?

恐怕不止一个。 Spark 开发人员认真努力提供稳定的公共(public) API。实现的低细节，包括 guest 语言如何与 JVM 通信的方式，根本不是契约(Contract)的一部分。它可以在任何时候完全重写，而不会对用户产生任何负面影响。如果您决定使用它并且存在向后不兼容的更改，那么您就只能靠自己了。

保持内部隐私可以减少维护和支持软件的工作量。您根本不必为用户滥用这些的所有可能方式而烦恼。

a better way to make such a cross-language library

如果不了解更多关于您的用例，就很难说。我看到至少三个选项:

对于初学者来说，R 只提供了一个弱访问控制机制。如果 API 的任何部分是内部的，您始终可以使用 :::访问它的功能。正如聪明人所说:

It is typically a design mistake to use ::: in your code since the corresponding object has probably been kept internal for a good reason.

但可以肯定的是，它比修改 Spark 源代码要好得多。作为奖励，它清楚地标记了您的代码中特别脆弱和可能不稳定的部分。

如果您只想创建 DataFrame，最简单的方法就是使用原始 SQL。它干净、便携、不需要编译、打包并且简单地工作。假设您将如下查询字符串存储在名为 q 的变量中

CREATE TEMPORARY TABLE foo
USING org.apache.spark.sql.jdbc
OPTIONS (
    url "jdbc:postgresql://localhost/test",
    dbtable "public.foo",
    driver "org.postgresql.Driver"
)

它可以在 R 中使用:

sql(sqlContext, q)
fooDF <- sql(sqlContext, "SELECT * FROM foo")

Python:

sqlContext.sql(q)
fooDF = sqlContext.sql("SELECT * FROM foo")

斯卡拉:

sqlContext.sql(q)
val fooDF = sqlContext.sql("SELECT * FROM foo")

或直接在 Spark SQL 中。

最后你可以使用Spark Data Sources API实现一致且受支持的跨平台访问。

在这三个中，我更喜欢原始 SQL，其次是用于复杂情况的数据源 API，并将内部作为最后的手段。

编辑 (2016-08-04):

如果您对 JVM 的低级访问感兴趣，可以使用相对较新的包 rstudio/sparkapi它公开了内部 SparkR RPC 协议(protocol)。很难预测它将如何演变，因此使用它需要您自担风险。