从任务调用 Java/Scala 函数 [英] Calling Java/Scala function from a task

问题描述

背景

我最初的问题是为什么在 map 函数中使用 DecisionTreeModel.predict 会引发异常? 并且与

由于 Py4J 网关在驱动程序上运行，因此 Python 解释器无法通过套接字与 JVM 工作线程进行通信(参见例如 PythonRDD/rdd.py).

理论上可以为每个工人创建一个单独的 Py4J 网关，但实际上它不太可能有用.忽略可靠性等问题 Py4J 根本就不是为执行数据密集型任务而设计的.

有什么解决方法吗?

1. 使用 Spark SQL 数据源 API 来包装 JVM 代码.

   优点:受支持、高级别的，不需要访问内部 PySpark API

   缺点:相对冗长且没有很好的文档记录，主要限于输入数据

2. 使用 Scala UDF 对 DataFrame 进行操作.

   优点:易于实现(参见 Spark:如何使用 Scala 或 Java 用户定义函数映射 Python?)，如果数据已经存储在 DataFrame 中，则 Python 和 Scala 之间没有数据转换，对 Py4J 的访问最少

   缺点:需要访问 Py4J 网关和内部方法，仅限于 Spark SQL，难以调试，不支持

3. 以类似于在 MLlib 中完成的方式创建高级 Scala 界面.

   优点:灵活，能够执行任意复杂的代码.它可以直接在 RDD 上执行(参见例如 MLlib 模型包装器)或使用 DataFrames(参见 如何在 Pyspark 中使用 Scala 类).后一种解决方案似乎更友好，因为所有 ser-de 细节都已由现有 API 处理.

   缺点:低级，需要数据转换，与UDF一样需要访问Py4J和内部API，不支持

   可以在使用 Scala 转换 PySpark RDD

中找到一些基本示例

4. 使用外部工作流管理工具在 Python 和 Scala/Java 作业之间切换并将数据传递到 DFS.

   优点:易于实现，对代码本身的改动最小

   缺点:读取/写入数据的成本(Alluxio?)

5. 使用共享的 SQLContext(参见例如 Apache Zeppelin或 Livy) 使用注册的临时表在来宾语言之间传递数据.

   优点:非常适合交互式分析

   缺点:批处理作业 (Zeppelin) 的影响不大，或者可能需要额外的编排 (Livy)

<小时>

1. 约书亚·罗森.(2014 年 8 月 4 日)PySpark 内部结构.检索自 https://cwiki.apache.org/confluence/display/SPARK/PySpark+内部

Background

My original question here was Why using DecisionTreeModel.predict inside map function raises an exception? and is related to How to generate tuples of (original lable, predicted label) on Spark with MLlib?

When we use Scala API a recommended way of getting predictions for RDD[LabeledPoint] using DecisionTreeModel is to simply map over RDD:

val labelAndPreds = testData.map { point =>
  val prediction = model.predict(point.features)
  (point.label, prediction)
}

Unfortunately similar approach in PySpark doesn't work so well:

labelsAndPredictions = testData.map(
    lambda lp: (lp.label, model.predict(lp.features))
labelsAndPredictions.first()

Exception: It appears that you are attempting to reference SparkContext from a broadcast variable, action, or transforamtion. SparkContext can only be used on the driver, not in code that it run on workers. For more information, see SPARK-5063.

Instead of that official documentation recommends something like this:

predictions = model.predict(testData.map(lambda x: x.features))
labelsAndPredictions = testData.map(lambda lp: lp.label).zip(predictions)

So what is going on here? There is no broadcast variable here and Scala API defines predict as follows:

/**
 * Predict values for a single data point using the model trained.
 *
 * @param features array representing a single data point
 * @return Double prediction from the trained model
 */
def predict(features: Vector): Double = {
  topNode.predict(features)
}

/**
 * Predict values for the given data set using the model trained.
 *
 * @param features RDD representing data points to be predicted
 * @return RDD of predictions for each of the given data points
 */
def predict(features: RDD[Vector]): RDD[Double] = {
  features.map(x => predict(x))
}

so at least at the first glance calling from action or transformation is not a problem since prediction seems to be a local operation.

Explanation

After some digging I figured out that the source of the problem is a JavaModelWrapper.call method invoked from DecisionTreeModel.predict. It access SparkContext which is required to call Java function:

callJavaFunc(self._sc, getattr(self._java_model, name), *a)

Question

In case of DecisionTreeModel.predict there is a recommended workaround and all the required code is already a part of the Scala API but is there any elegant way to handle problem like this in general?

Only solutions I can think of right now are rather heavyweight:

• pushing everything down to JVM either by extending Spark classes through Implicit Conversions or adding some kind of wrappers
• using Py4j gateway directly

解决方案

Communication using default Py4J gateway is simply not possible. To understand why we have to take a look at the following diagram from the PySpark Internals document [1]:

Since Py4J gateway runs on the driver it is not accessible to Python interpreters which communicate with JVM workers through sockets (See for example PythonRDD / rdd.py).

Theoretically it could be possible to create a separate Py4J gateway for each worker but in practice it is unlikely to be useful. Ignoring issues like reliability Py4J is simply not designed to perform data intensive tasks.

Are there any workarounds?

1. Using Spark SQL Data Sources API to wrap JVM code.

   Pros: Supported, high level, doesn't require access to the internal PySpark API

   Cons: Relatively verbose and not very well documented, limited mostly to the input data

2. Operating on DataFrames using Scala UDFs.

   Pros: Easy to implement (see Spark: How to map Python with Scala or Java User Defined Functions?), no data conversion between Python and Scala if data is already stored in a DataFrame, minimal access to Py4J

   Cons: Requires access to Py4J gateway and internal methods, limited to Spark SQL, hard to debug, not supported

3. Creating high level Scala interface in a similar way how it is done in MLlib.

   Pros: Flexible, ability to execute arbitrary complex code. It can be don either directly on RDD (see for example MLlib model wrappers) or with DataFrames (see How to use a Scala class inside Pyspark). The latter solution seems to be much more friendly since all ser-de details are already handled by existing API.

   Cons: Low level, required data conversion, same as UDFs requires access to Py4J and internal API, not supported

   Some basic examples can be found in Transforming PySpark RDD with Scala

4. Using external workflow management tool to switch between Python and Scala / Java jobs and passing data to a DFS.

   Pros: Easy to implement, minimal changes to the code itself

   Cons: Cost of reading / writing data (Alluxio?)

5. Using shared SQLContext (see for example Apache Zeppelin or Livy) to pass data between guest languages using registered temporary tables.

   Pros: Well suited for interactive analysis

   Cons: Not so much for batch jobs (Zeppelin) or may require additional orchestration (Livy)

---

1. Joshua Rosen. (2014, August 04) PySpark Internals. Retrieved from https://cwiki.apache.org/confluence/display/SPARK/PySpark+Internals

这篇关于从任务调用 Java/Scala 函数的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！