Wie die Rückkehr ein "Tupel-Typ" in einer UDF in PySpark?

Alle Datentypen in pyspark.sql.Typen werden:

__all__ = [
    "DataType", "NullType", "StringType", "BinaryType", "BooleanType", "DateType",
    "TimestampType", "DecimalType", "DoubleType", "FloatType", "ByteType", "IntegerType",
    "LongType", "ShortType", "ArrayType", "MapType", "StructField", "StructType"]

Ich zu schreiben, ein UDF (in pyspark) gibt ein array von Tupeln. Was muss ich geben, das zweite argument ist der Typ des Rückgabewert der udf-Methode? Es würde etwas sein, auf den Linien der ArrayType(TupleType())...

Deine Titel-Frage scheint nicht zu passen den Körper. Nicht in der Dokumentation erzählen Sie, wie man einen return-Wert "- container Typ der anderen Art"?
Ich habe den Titel geändert. Hoffentlich ist es repräsentativ für den Körper jetzt.

InformationsquelleAutor kamalbanga | 2016-04-25

21

Gibt es keine solche Sache wie eine TupleType im Spark. Produkt-Arten sind vertreten, wie structs mit Feldern bestimmten Typs. Zum Beispiel, wenn Sie zurückkehren möchten, ein array mit Paaren (integer, string) verwenden Sie schema wie dieses:
```
from pyspark.sql.types import *

schema = ArrayType(StructType([
    StructField("char", StringType(), False),
    StructField("count", IntegerType(), False)
]))
```
Beispiel:
```
from pyspark.sql.functions import udf
from collections import Counter

char_count_udf = udf(
    lambda s: Counter(s).most_common(),
    schema
)

df = sc.parallelize([(1, "foo"), (2, "bar")]).toDF(["id", "value"])

df.select("*", char_count_udf(df["value"])).show(2, False)

## +---+-----+-------------------------+
## |id |value|PythonUDF#<lambda>(value)|
## +---+-----+-------------------------+
## |1  |foo  |[[o,2], [f,1]]           |
## |2  |bar  |[[r,1], [a,1], [b,1]]    |
## +---+-----+-------------------------+
```
- Deine Antwort-ist die Arbeit, aber mein Fall ist ein bisschen komplexer. Meine Daten zurück, ist der Typ [('a1', [('b1', 1), ('b2', 2)]), ('a2', [('b1', 1), ('b2', 2)])] und deshalb mache ich einen Typ wie ArrayType(StructType([StructField("date", StringType(), False), ArrayType(StructType([StructField("hashId", StringType(), False), StructField("TimeSpent-Front", FloatType(), False), StructField("TimeSpent-Back", FloatType(), False)]))])) gibt 'ArrayType' - Objekt hat kein Attribut 'name'...
- StructType erfordert eine Sequenz von StructFields daher können Sie nicht verwenden ArrayTypes allein. Sie müssen StructField welche Geschäfte ArrayType. Auch Wort der Beratung - wenn Sie finden, sich selbst Strukturen zu schaffen, wie dies sollten Sie wohl überdenken Daten-Modell. Tief geschachtelte Strukturen sind schwer zu handhaben, ohne UDFs und Python-UDFs sind weit aus effizienter.
- Wie kann ich nur angeben schema in udf um eine Liste . F. udf(lambda start_date, end_date : [0,1] wenn start_date < end_date anderes [1]).
InformationsquelleAutor zero323

Stackoverflow hält lenken mich zu dieser Frage, also ich denke, ich werde einige Infos hier.

Rückkehr einfache Typen von UDF:

from pyspark.sql.types import *
from pyspark.sql import functions as F

def get_df():
  d = [(0.0, 0.0), (0.0, 3.0), (1.0, 6.0), (1.0, 9.0)]
  df = sqlContext.createDataFrame(d, ['x', 'y'])
  return df

df = get_df()
df.show()

# +---+---+
# |  x|  y|
# +---+---+
# |0.0|0.0|
# |0.0|3.0|
# |1.0|6.0|
# |1.0|9.0|
# +---+---+

func = udf(lambda x: str(x), StringType())
df = df.withColumn('y_str', func('y'))

func = udf(lambda x: int(x), IntegerType())
df = df.withColumn('y_int', func('y'))

df.show()

# +---+---+-----+-----+
# |  x|  y|y_str|y_int|
# +---+---+-----+-----+
# |0.0|0.0|  0.0|    0|
# |0.0|3.0|  3.0|    3|
# |1.0|6.0|  6.0|    6|
# |1.0|9.0|  9.0|    9|
# +---+---+-----+-----+

df.printSchema()

# root
#  |-- x: double (nullable = true)
#  |-- y: double (nullable = true)
#  |-- y_str: string (nullable = true)
#  |-- y_int: integer (nullable = true)

Als Ganzzahlen sind nicht genug:

df = get_df()

func = udf(lambda x: [0]*int(x), ArrayType(IntegerType()))
df = df.withColumn('list', func('y'))

func = udf(lambda x: {float(y): str(y) for y in range(int(x))}, 
           MapType(FloatType(), StringType()))
df = df.withColumn('map', func('y'))

df.show()
# +---+---+--------------------+--------------------+
# |  x|  y|                list|                 map|
# +---+---+--------------------+--------------------+
# |0.0|0.0|                  []|               Map()|
# |0.0|3.0|           [0, 0, 0]|Map(2.0 -> 2, 0.0...|
# |1.0|6.0|  [0, 0, 0, 0, 0, 0]|Map(0.0 -> 0, 5.0...|
# |1.0|9.0|[0, 0, 0, 0, 0, 0...|Map(0.0 -> 0, 5.0...|
# +---+---+--------------------+--------------------+

df.printSchema()
# root
#  |-- x: double (nullable = true)
#  |-- y: double (nullable = true)
#  |-- list: array (nullable = true)
#  |    |-- element: integer (containsNull = true)
#  |-- map: map (nullable = true)
#  |    |-- key: float
#  |    |-- value: string (valueContainsNull = true)

Rückkehr komplexe Datentypen von UDF:

df = get_df()
df = df.groupBy('x').agg(F.collect_list('y').alias('y[]'))
df.show()

# +---+----------+
# |  x|       y[]|
# +---+----------+
# |0.0|[0.0, 3.0]|
# |1.0|[9.0, 6.0]|
# +---+----------+

schema = StructType([
    StructField("min", FloatType(), True),
    StructField("size", IntegerType(), True),
    StructField("edges",  ArrayType(FloatType()), True),
    StructField("val_to_index",  MapType(FloatType(), IntegerType()), True)
    # StructField('insanity', StructType([StructField("min_", FloatType(), True), StructField("size_", IntegerType(), True)]))

])

def func(values):
  mn = min(values)
  size = len(values)
  lst = sorted(values)[::-1]
  val_to_index = {x: i for i, x in enumerate(values)}
  return (mn, size, lst, val_to_index)

func = udf(func, schema)
dff = df.select('*', func('y[]').alias('complex_type'))
dff.show(10, False)

# +---+----------+------------------------------------------------------+
# |x  |y[]       |complex_type                                          |
# +---+----------+------------------------------------------------------+
# |0.0|[0.0, 3.0]|[0.0,2,WrappedArray(3.0, 0.0),Map(0.0 -> 0, 3.0 -> 1)]|
# |1.0|[6.0, 9.0]|[6.0,2,WrappedArray(9.0, 6.0),Map(9.0 -> 1, 6.0 -> 0)]|
# +---+----------+------------------------------------------------------+

dff.printSchema()

# +---+----------+------------------------------------------------------+
# |x  |y[]       |complex_type                                          |
# +---+----------+------------------------------------------------------+
# |0.0|[0.0, 3.0]|[0.0,2,WrappedArray(3.0, 0.0),Map(0.0 -> 0, 3.0 -> 1)]|
# |1.0|[6.0, 9.0]|[6.0,2,WrappedArray(9.0, 6.0),Map(9.0 -> 1, 6.0 -> 0)]|
# +---+----------+------------------------------------------------------+

Die übergabe mehrerer Argumente einer UDF:

df = get_df()
func = udf(lambda arr: arr[0]*arr[1],FloatType())
df = df.withColumn('x*y', func(F.array('x', 'y')))

    # +---+---+---+
    # |  x|  y|x*y|
    # +---+---+---+
    # |0.0|0.0|0.0|
    # |0.0|3.0|0.0|
    # |1.0|6.0|6.0|
    # |1.0|9.0|9.0|
    # +---+---+---+

Den code rein für demo-Zwecke, vor allem der transformation stehen im Spark-code, und würde die Ausbeute wesentlich bessere performance.
Wie @zero323 in den Kommentar oben, UDFs sollten generell vermieden werden, in pyspark; Rückkehr komplexe Typen sollten Sie darüber nachdenken, vereinfachen Sie Ihre Logik.

InformationsquelleAutor y.selivonchyk

Für die scala-version anstelle von python.
version 2.4

import org.apache.spark.sql.types._

val testschema : StructType = StructType(
    StructField("number", IntegerType) ::
    StructField("Array",  ArrayType(StructType(StructField("cnt_rnk", IntegerType) :: StructField("comp", StringType) :: Nil))) :: 
    StructField("comp", StringType):: Nil)

Die Baum-Struktur sieht wie folgt aus.

testschema.printTreeString
root
 |-- number: integer (nullable = true)
 |-- Array: array (nullable = true)
 |    |-- element: struct (containsNull = true)
 |    |    |-- cnt_rnk: integer (nullable = true)
 |    |    |-- corp_id: string (nullable = true)
 |-- comp: string (nullable = true)

InformationsquelleAutor Achyuth

Schreibe einen Kommentar

Du musst angemeldet sein, um einen Kommentar abzugeben.