例如，假设我的团队选择 Python 作为使用 Spark 开发的引用语言。但后来出于性能原因，我们希望开发特定的 Scala 或 Java 特定库，以便将它们与我们的 Python 代码映射(类似于具有 Scala 或 Java 骨架的 Python stub )。
您不认为有可能将新的自定义 Python 方法与一些 Scala 或 Java 用户定义函数进行接口(interface)吗？
最佳答案
Spark 2.1+
您可以使用 SQLContext.registerJavaFunction :
Register a java UDF so it can be used in SQL statements.
需要 name、Java 类的完全限定名称和可选的返回类型。不幸的是，目前它只能在 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 的包装器。
假设我想重用 GroupConcat 作为对 SPARK SQL replacement for mysql GROUP_CONCAT aggregate function 的回答而创建的 UDAF它位于 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 - Spark : How to map Python with Scala or Java User Defined Functions?，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/33233737/