Wywołanie funkcji Java / Scala z zadania

Tło

Moje pierwotne pytanie brzmiało dlaczego użycie DecisionTreeModel.predict wewnątrz funkcji mapy powoduje wyjątek? i jest związane z Jak wygenerować krotki (oryginalna etykieta, przewidywana etykieta) na Spark za pomocą MLlib?

Kiedy używamy Scala API zalecanym sposobem uzyskiwania prognoz dla RDD[LabeledPoint] za pomocą DecisionTreeModel jest po prostu mapowanie na RDD:

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

Niestety podobne podejście w PySpark nie działa tak dobrze:

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

Wyjątek: wygląda na to, że próbujesz odwołać się do SparkContext ze zmiennej broadcast, akcji lub transforamtion. SparkContext może być używany tylko na sterowniku, a nie w kodzie uruchamianym na workerach. Aby uzyskać więcej informacji, zobacz SPARK-5063.

Zamiast tego oficjalna dokumentacja zaleca coś takiego:

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

/**
 * 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))
}

Więc przynajmniej na pierwszy rzut oka wywołanie z akcji lub transformacji nie jest problemem, ponieważ przewidywanie wydaje się być operacją lokalną.

Wyjaśnienie

Po pewnym kopaniu zorientowałem się, że źródłem problemu jest JavaModelWrapper.call metoda przywołana z DecisionTreeModel.predict . It access SparkContext który jest wymagany do wywołania funkcji Java:

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

Pytanie

W przypadku DecisionTreeModel.predict istnieje zalecane obejście i cały wymagany kod jest już częścią Scala API, ale czy jest jakiś elegancki sposób, aby poradzić sobie z takim problemem w ogóle?

Jedyne rozwiązania, o których teraz myślę, są raczej ciężkie:

• przepychanie wszystkiego do JVM poprzez rozszerzanie klas Spark poprzez ukryte konwersje lub dodawanie pewnego rodzaju opakowań
• korzystanie bezpośrednio z bramy Py4j