Spark: How to map Python mit Scala oder Java-User Defined Functions?

Lassen Sie uns sagen zum Beispiel, dass mein team ausgewählt hat Python als Referenz Sprache zu entwickeln, mit Funke. Aber später aus Gründen der Leistung, die wir entwickeln möchten, bestimmte Scala oder Java-spezifische Bibliotheken, um die Karte mit unseren Python-code (etwas, das ähnlich wie Python stubs mit Scala oder Java-skeletons).

Nicht denken Sie, ist es möglich, interface-neue angepasste Python-Methoden mit unter der Haube einige Scala-oder Java-User Defined Functions ?

InformationsquelleAutor prossblad | 2015-10-20
  1. 26

    Funke 2.1+

    Können Sie SQLContext.registerJavaFunction:

    Registrieren Sie einen java-UDF-so kann es verwendet werden, die in SQL-Anweisungen.

    erfordert eine name voll qualifizierte name der Java-Klasse und optionalen Rückgabetyp. Leider für jetzt kann es nur verwendet werden, in SQL-Anweisungen (oder mit expr /selectExpr) und benötigt eine 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 hat:

    Ich würde nicht so weit gehen zu sagen, dass es unterstützt wird, aber es ist sicherlich möglich. Alle SQL-Funktionen zur Verfügung, die derzeit in PySpark sind einfach nur ein Wrapper für Scala-API.

    Vermuten lässt möchte ich wiederverwenden GroupConcat UDAF habe ich erstellt als eine Antwort auf SPARK SQL Ersatz für mysql Aggregat-Funktion GROUP_CONCAT und es befindet sich in einem Paket 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|
## +---+---------+

    Es ist viel zu viel führende Unterstriche für meinen Geschmack, aber als Sie kann sehen, es kann getan werden.

    Bezug auf:

    • Ich mache Folgendes, aber jedes mal, wenn ich Begegnung "py4j.Protokoll.Py4JError": Kom.Beispiel.udf.GroupConcat.gelten nicht in der JVM. Mein Paket ist "com.Beispiel.udf"
    • Dies bedeutet in der Regel falsch CLASSPATH
    • Ich habe eine jar, die enum-Konstanten und UDF. Zum ändern dieser code zu verwenden?
    • Bin ich etwas fehlt, wie registerJavaFunction weiß, wo zu finden, Ihre UDF... Könnten Sie näher auf die Verzeichnis-Struktur hier? Haben Sie sbt clean assembly Ihre scalaVersion :=... (build.sbt??) und package com.example.spark.udfs... (src/main/scala??) Dateien aus einem anderen Verzeichnis?anderswo?
    • Es ist erwähnenswert, dass, sollten Sie wirklich überprüfen Sie zuerst repo1.maven.org/maven2/org/apache/spark um sicher zu sein Ihre Scala und Spark-Versionen kompatibel sind in Erster Linie... ich verbrachte einen ganzen Tag (mein Erster Tag mit sbt ?), die versuchen, zu scalaVersion := "2.12.7" Arbeit mit sparkVersion = "2.3.1", aber Scala 2.12+ nur kompatibel mit Spark 2.4+ (oder so entnehme ich)
    InformationsquelleAutor zero323
Schreibe einen Kommentar