Spark: Jak zmapować Pythona za pomocą funkcji zdefiniowanych przez użytkownika Scala lub Java?

Question

Załóżmy na przykład, że mój zespół wybrał Python jako język odniesienia do opracowania ze Spark. Ale później ze względu na wydajność, chcielibyśmy opracować specyficzne biblioteki librairies w Scali lub Javie, aby zmapować je za pomocą naszego kodu Pythona (coś podobnego do kodu Pythona ze szkieletami Scala lub Java).

Czy nie sądzisz, że możliwe jest połączenie nowych, spersonalizowanych metod w Pythonie z niektórymi funkcjami Scala lub Java User Defined?

Answer 1

Spark 2.1+

Można użyć SQLContext.registerJavaFunction:

Zarejestruj UDF java dzięki czemu może być stosowany w SQL.

Wymagana jest pełna nazwa klasy Java i opcjonalny typ powrotu: name. Niestety na razie może on być używany wyłącznie w instrukcjach SQL (lub z expr/selectExpr) i wymaga 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| 
## +------+ 

Wersja indpendent:

nie będę posunąć się nawet do stwierdzenia, że ​​jest obsługiwany, ale jest to z pewnością możliwe. Wszystkie funkcje SQL dostępne obecnie w PySpark to po prostu wrappery wokół API Scala.

Załóżmy chcę ponowne GroupConcat UDAF Utworzona jako odpowiedź na SPARK SQL replacement for mysql GROUP_CONCAT aggregate function i znajduje się w opakowaniu 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| 
## +---+---------+ 

Jest zbyt dużo wiodącym podkreślenia jak na mój gust, ale jak ty widzę, że można to zrobić.

związane z:

• Calling Java/Scala function from a task
• How to use a Scala class inside Pyspark
• Transforming PySpark RDD with Scala