How to use a Scala class inside Pyspark

user_1689706 0
Viewed 19140

I've been searching for a while if there is any way to use a Scala class in Pyspark, and I haven't found any documentation nor guide about this subject.

Let's say I create a simple class in Scala that uses some libraries of apache-spark, something like:

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

    df.select(col(column))
  }
}
  • Is there any possible way to use this class in Pyspark?
  • Is it too tough?
  • Do I have to create a .py file?
  • Is there any guide that shows how to do that?

By the way I also looked at the spark code and I felt a bit lost, and I was incapable of replicating their functionality for my own purpose.

pyspark apache-spark-sql apache-spark scala python
2 Answers

As an update to @zero323's answer, given that Spark's APIs have evolved over the last six years, a recipe that works in Spark-3.2 is as follows:

  1. Compile your Scala code into a JAR file (e.g. using sbt assembly)
  2. Include the JAR file in the --jars argument to spark-submit together with any --py-files arguments needed for local package definitions
  3. Extract the JVM instance within Python:
jvm = spark._jvm
  1. Extract a Java representation of the SparkSession:
jSess = spark._jsparkSession
  1. Extract the Java representation of the PySpark DataFrame:
jdf = df._jdf
  1. Create a new instance of SimpleClass:
simpleObject = jvm.com.example.SimpleClass(jSess, jdf, "v")
  1. Call the exe method and convert its output into a PySpark DataFrame:
from pyspark.sql import DataFrame

result = DataFrame(simpleObject.exe(), spark)

If you need to pass additional parameters, such as a Python dictionary, PySpark may automatically convert them into corresponding Java types they before emerge into your Scala methods. Scala provides the JavaConverters package to help with translating this into more natural Scala datatypes. For example, a Python dictionary could be passed into a Scala method and immediately converted from a Java HashMap into a Scala (mutable) Map:

def processDict(spark: SparkSession, jparams: java.util.Map[String, Any]) {
  import scala.collection.JavaConverters._
  val params = jparams.asScala
  ...
}
user_92149850
Related