Как использовать класс Scala внутри Pyspark

Я искал какое-то время, если есть какой-либо способ использования класса Scala в Pyspark, и я не нашел никакой документации или руководства по этому вопросу.

Скажем, я создаю простой класс в Scala, который использует некоторые библиотеки apache-spark, что-то вроде:

class SimpleClass(sqlContext: SQLContext, df: DataFrame, column: String) {
  def exe(): DataFrame = {
    import sqlContext.implicits._

    df.select(col(column))
  }
}
  • Можно ли использовать этот класс в Pyspark?
  • Это слишком сложно?
  • Должен ли я создать файл .py?
  • Есть ли какой-нибудь справочник, который показывает, как это сделать?

Кстати, я также посмотрел код spark, и я чувствовал себя немного потерянным, и я не мог воспроизвести их функциональность для своей собственной цели.

Ответы

Ответ 1

Да, возможно, это может быть далеко не тривиально. Как правило, вам нужна оболочка Java (дружественная), поэтому вам не нужно иметь дело с Scala функциями, которые не могут быть легко выражены с помощью простой Java и, как результат, не хорошо работают с шлюзом Py4J.

Предполагая, что ваш класс является int пакетом com.example и имеет Python DataFrame, называемый df

df = ... # Python DataFrame

вам нужно:

  • Создайте банку, используя ваш любимый инструмент сборки.

  • Включите его в путь класса драйвера, например, используя аргумент --driver-class-path для оболочки PySpark/spark-submit. В зависимости от точного кода вам может потребоваться передать его, используя --jars, а также

  • Извлечь экземпляр JVM из экземпляра Python SparkContext:

    jvm = sc._jvm

  • Извлеките Scala SQLContext из экземпляра SQLContext:

    ssqlContext = sqlContext._ssql_ctx

  • Извлеките Java DataFrame из df:

    jdf = df._jdf

  • Создайте новый экземпляр SimpleClass:

    simpleObject = jvm.com.example.SimpleClass(ssqlContext, jdf, "v")

  • Вызвать exe метод и обернуть результат с помощью Python DataFrame:

    from pyspark.sql import DataFrame

DataFrame(simpleObject.exe(), ssqlContext)

Результат должен быть действительным PySpark DataFrame. Вы можете, конечно, объединить все шаги в один вызов.

Важно. Этот подход возможен, только если код Python выполняется исключительно на драйвере. Он не может использоваться внутри действия или преобразования Python. См. Как использовать функцию Java/ Scala из действия или преобразования? для деталей.