Spark：如何将 Python 与 Scala 或 Java 用户定义函数映射？

问题描述：

假设我的团队已选择 Python 作为使用 Spark 进行开发的参考语言。但后来出于性能原因，我们想开发特定的 Scala 或 Java 特定库，以便将它们与我们的 Python 代码进行映射（类似于带有 Scala 或 Java 骨架的 Python 存根）。

您不觉得有可能将新的定制 Python 方法与某些 Scala 或 Java 用户定义函数进行交互吗？

解决方案 1：

Spark 2.1+

您可以使用SQLContext.registerJavaFunction：

注册一个 java UDF，以便它可以在 SQL 语句中使用。

它需要一个nameJava 类的完全限定名称和可选的返回类型。遗憾的是，目前它只能在 SQL 语句中使用（或与expr/一起使用selectExpr），并且需要一个 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|
## +------+

版本独立：

我不会说它受支持，但肯定是可能的。PySpark 中当前可用的所有 SQL 函数都只是 Scala API 的包装器。

假设我想重用我创建的 UDAF 作为SPARK SQL 替换 mysql GROUP_CONCAT 聚合函数的GroupConcat答案，它位于一个包中：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|
## +---+---------+

对于我的口味来说，前导下划线太多了，但正如你所见，这是可以做到的。

相关：

• 从任务调用 Java/Scala 函数

• 如何在 Pyspark 中使用 Scala 类

• 使用 Scala 转换 PySpark RDD