Spark:如何使用Scala或Java用户定义函数映射Python？

Question

比方说,我的团队选择Python作为Spark开发的参考语言.但是后来由于性能原因,我们希望开发特定的Scala或Java特定的库,以便使用我们的Python代码(类似于使用Scala或Java骨架的Python存根)进行映射.

难道您不认为是否可以将新的自定义Python方法与一些Scala或Java用户定义函数联系起来？

Answer 1

Spark 2.1+

你可以使用SQLContext.registerJavaFunction:

注册java UDF,以便可以在SQL语句中使用它.

这需要一个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我创建的UDAF作为SPARK SQL替换mysql GROUP_CONCAT聚合函数的答案,它位于一个包中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