Apache Spark — Weisen Sie das Ergebnis der UDF auf mehrere dataframe Spalten

Ich bin mit pyspark, laden eine große csv-Datei in einem dataframe mit Funken -, csv -, und als pre-processing Schritt, den ich anwenden müssen, eine Vielzahl von Operationen, um die verfügbaren Daten in einer der Spalten (enthält einen json-string). Zurückgeben, X-Werte, von denen jeder braucht, um gespeichert werden in einer eigenen Spalte.

Dass die Funktionalität implementiert werden, die in einer UDF. Allerdings bin ich mir nicht sicher, wie das zurückgeben einer Liste von Werten aus, die UDF und führen diese in einzelnen Spalten. Unten ist ein einfaches Beispiel:

(...)
from pyspark.sql.functions import udf
def udf_test(n):
    return [n/2, n%2]

test_udf=udf(udf_test)


df.select('amount','trans_date').withColumn("test", test_udf("amount")).show(4)

Produziert die folgenden:

+------+----------+--------------------+
|amount|trans_date|                test|
+------+----------+--------------------+
|  28.0|2016-02-07|         [14.0, 0.0]|
| 31.01|2016-02-07|[15.5050001144409...|
| 13.41|2016-02-04|[6.70499992370605...|
| 307.7|2015-02-17|[153.850006103515...|
| 22.09|2016-02-05|[11.0450000762939...|
+------+----------+--------------------+
only showing top 5 rows

Was wäre der beste Weg, um die beiden (in diesem Beispiel) Werte zurückgegeben, die von der udf auf separate Spalten? Jetzt sind Sie als Zeichenfolgen eingegeben:

df.select('amount','trans_date').withColumn("test", test_udf("amount")).printSchema()

root
 |-- amount: float (nullable = true)
 |-- trans_date: string (nullable = true)
 |-- test: string (nullable = true)
InformationsquelleAutor Everaldo Aguiar | 2016-02-10
  1. 58

    Ist es nicht möglich, mehrere top-level-Spalten aus einer einzelnen UDF aufrufen, aber Sie können erstellen Sie eine neue struct. Es erfordert eine UDF mit angegeben returnType:

    from pyspark.sql.functions import udf
from pyspark.sql.types import *

schema = StructType([
    StructField("foo", FloatType(), False),
    StructField("bar", FloatType(), False)
])

def udf_test(n):
    return (n / 2, n % 2) if n and n != 0.0 else (float('nan'), float('nan'))

test_udf = udf(udf_test, schema)
df = sc.parallelize([(1, 2.0), (2, 3.0)]).toDF(["x", "y"])

foobars = df.select(test_udf("y").alias("foobar"))
foobars.printSchema()
## root
##  |-- foobar: struct (nullable = true)
##  |    |-- foo: float (nullable = false)
##  |    |-- bar: float (nullable = false)

    Sie weiter reduzieren Sie das schema mit einfachen select:

    foobars.select("foobar.foo", "foobar.bar").show()
## +---+---+
## |foo|bar|
## +---+---+
## |1.0|0.0|
## |1.5|1.0|
## +---+---+

    Siehe auch Leiten mehrerer Spalten aus einer einzelnen Spalte in einer Spark-DataFrame

    • Fantastisch! Dies funktioniert sehr gut für das, was ich brauchte. Ich war auf dem besten Weg dorthin, aber war die Fütterung der StructType schema falsch auf die udf, die verursacht wurde meine neue Spalte am Ende als StringType statt. Dank einem Haufen!
    • Danke!! Das war genau das, was ich suchte. 🙂
    • Sie können auch foobars.select("foobar.*") anstelle von benennen Sie jede Spalte individuell.
    • Sie können auch "mischen" sich die beiden ursprünglichen Spalten und diese von der UDF, die in einem zwei-Schritt-Prozess: df.select("x", test_udf("y").alias("foobar")).select("x", "foobar.*")
    InformationsquelleAutor zero323
Schreibe einen Kommentar