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

Question

Я искал какое-то время, если есть способ использовать класс 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 код, и я чувствовал себя немного потеряли, и я был не в состоянии копировать их функциональность для моей собственной цели.

Answer 1

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

Предполагая, что ваш класс ИНТ пакет com.example и есть Python DataFrame под названием df

df = ... # Python DataFrame 

вам придется:

1. Построить банку с помощью your favorite build tool.

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

3. Извлечение экземпляра JVM из Python SparkContext Например:

   jvm = sc._jvm 
   

4. Экстракт Scala SQLContext из SQLContext Например:

   ssqlContext = sqlContext._ssql_ctx 
   

5. Выдержка Java DataFrame от df:

   jdf = df._jdf 
   

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

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

7. Позвоните exe метод и завернуть результат, используя Python DataFrame:

   from pyspark.sql import DataFrame 
   
   DataFrame(simpleObject.exe(), ssqlContext) 
   

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

Важный: Этот подход возможен, только если код Python выполняется исключительно на драйвере. Он не может использоваться внутри действия или преобразования Python. См. How to use Java/Scala function from an action or a transformation?.