Spark: How to map Python with Scala or Java User Defined Functions?

by

Spark 2.1+

You can use SQLContext.registerJavaFunction:

Register a java UDF so it can be used in SQL statements.

which requires a name, fully qualified name of Java class, and optional return type. Unfortunately for now it can be used only in SQL statements (or with expr / selectExpr) and requires a Java org.apache.spark.sql.api.java.UDF*:

scalaVersion := "2.11.8"

libraryDependencies ++= Seq(
  "org.apache.spark" %% "spark-sql" % "2.1.0"
)

package com.example.spark.udfs

import org.apache.spark.sql.api.java.UDF1

class addOne extends UDF1[Integer, Integer] {
  def call(x: Integer) = x + 1
} 

sqlContext.registerJavaFunction("add_one", "com.example.spark.udfs.addOne")
sqlContext.sql("SELECT add_one(1)").show()

## +------+
## |UDF(1)|
## +------+
## |     2|
## +------+

Version indpendent:

I wouldn’t go so far as to say it is supported but it is certainly possible. All SQL functions available currently in PySpark are simply a wrappers around Scala API.

Lets assume I want to reuse GroupConcat UDAF I’ve created as an answer to SPARK SQL replacement for mysql GROUP_CONCAT aggregate function and it is located in a package com.example.udaf:

from pyspark.sql.column import Column, _to_java_column, _to_seq
from pyspark.sql import Row

row = Row("k", "v")
df = sc.parallelize([
    row(1, "foo1"), row(1, "foo2"), row(2, "bar1"), row(2, "bar2")]).toDF()

def groupConcat(col):
    """Group and concatenate values for a given column

    >>> df = sqlContext.createDataFrame([(1, "foo"), (2, "bar")], ("k", "v"))
    >>> df.select(groupConcat("v").alias("vs"))
    [Row(vs=u'foo,bar')]
    """
    sc = SparkContext._active_spark_context
    # It is possible to use java_import to avoid full package path
    _groupConcat = sc._jvm.com.example.udaf.GroupConcat.apply
    # Converting to Seq to match apply(exprs: Column*)
    return Column(_groupConcat(_to_seq(sc, [col], _to_java_column)))

df.groupBy("k").agg(groupConcat("v").alias("vs")).show()

## +---+---------+
## |  k|       vs|
## +---+---------+
## |  1|foo1,foo2|
## |  2|bar1,bar2|
## +---+---------+

There is far too much leading underscores for my taste but as you can see it can be done.

Related to:

More Related Contents:

  1. Pyspark: Exception: Java gateway process exited before sending the driver its port number
  2. Spark Error – Unsupported class file major version
  3. What are the Spark transformations that causes a Shuffle?
  4. Running custom Java class in PySpark
  5. Resolving dependency problems in Apache Spark
  6. Add JAR files to a Spark job – spark-submit
  7. Calling Java/Scala function from a task
  8. How to use JDBC source to write and read data in (Py)Spark?
  9. How to flatten a struct in a Spark dataframe?
  10. How to turn off INFO logging in Spark?
  11. Spark read file from S3 using sc.textFile (“s3n://…)
  12. Spark Strutured Streaming automatically converts timestamp to local time
  13. How can I update a broadcast variable in spark streaming?
  14. Which programming languages can I use on Android Dalvik?
  15. Apache Spark – foreach Vs foreachPartition When to use What?
  16. Matrix Multiplication in Apache Spark [closed]
  17. How to use a Scala class inside Pyspark
  18. PySpark: java.lang.OutofMemoryError: Java heap space
  19. Spark spark-submit –jars arguments wants comma list, how to declare a directory of jars?
  20. How to set timezone to UTC in Apache Spark?
  21. Is gzip format supported in Spark?
  22. Change output filename prefix for DataFrame.write()
  23. How can I force Spark to execute code?
  24. What is the equivalent to scala.util.Try in pyspark?
  25. How to save models from ML Pipeline to S3 or HDFS?
  26. Difference between SparkContext, JavaSparkContext, SQLContext, and SparkSession?
  27. Pyspark: Split multiple array columns into rows
  28. PySpark logging from the executor
  29. Equivalent of Scala dropWhile
  30. What is the best way to remove accents with Apache Spark dataframes in PySpark?

Leave a Comment