PySpark wierszami funkcją składu

Question

W uproszczonym przykładzie mam dataframe „DF” z kolumny „kol1, kolumna2” i chce obliczenie maksimum w rzędach po zastosowaniu funkcji każdej kolumny:

def f(x): 
    return (x+1) 

max_udf=udf(lambda x,y: max(x,y), IntegerType()) 
f_udf=udf(f, IntegerType()) 

df2=df.withColumn("result", max_udf(f_udf(df.col1),f_udf(df.col2))) 

Więc jeśli DF:

col1 col2 
1  2 
3  0 

Następnie

df2:

col1 col2 result 
1  2  3 
3  0  4 

Powyższe nie wydają się działać i produkuje „Nie można ocenić wyrażenia: PythonUDF # f ...”

jestem absolutnie pozytywne „f_udf” działa dobrze na moim stole, a głównym problemem jest to, ze max_udf.

Bez tworzenia dodatkowych kolumn lub używania podstawowej mapy/zmniejszenia, czy istnieje sposób na wykonanie powyższych czynności w całości przy użyciu ramek danych i udfs? Jak mam zmodyfikować "max_udf"?

Próbowałem również:

max_udf=udf(max, IntegerType()) 

która produkuje ten sam błąd.

Ja również potwierdziła, że ​​następujące utwory:

df2=(df.withColumn("temp1", f_udf(df.col1)) 
     .withColumn("temp2", f_udf(df.col2)) 

df2=df2.withColumn("result", max_udf(df2.temp1,df2.temp2)) 

Dlaczego jest to, że nie mogę zrobić to za jednym zamachem?

Chciałbym zobaczyć odpowiedź, która uogólnia każdą funkcję "f_udf" i "max_udf".

Answer 1

miałem podobny problem i znalazł rozwiązanie w odpowiedzi na this stackoverflow question

zdać wiele kolumn lub cały rząd do UDF użyć struct:

from pyspark.sql.functions import udf, struct 
from pyspark.sql.types import IntegerType 

df = sqlContext.createDataFrame([(None, None), (1, None), (None, 2)], ("a", "b")) 

count_empty_columns = udf(lambda row: len([x for x in row if x == None]), IntegerType()) 

new_df = df.withColumn("null_count", count_empty_columns(struct([df[x] for x in df.columns]))) 

new_df.show() 

zwrócone:

+----+----+----------+ 
| a| b|null_count| 
+----+----+----------+ 
|null|null|   2| 
| 1|null|   1| 
|null| 2|   1| 
+----+----+----------+ 

Answer 2

UserDefinedFunction rzuca błąd akceptując UDF jako ich argumenty.

Możesz zmodyfikować max_udf jak poniżej, aby działał.

df = sc.parallelize([(1, 2), (3, 0)]).toDF(["col1", "col2"]) 

max_udf = udf(lambda x, y: max(x + 1, y + 1), IntegerType()) 

df2 = df.withColumn("result", max_udf(df.col1, df.col2)) 

Albo

def f_udf(x): 
    return (x + 1) 

max_udf = udf(lambda x, y: max(x, y), IntegerType()) 
## f_udf=udf(f, IntegerType()) 

df2 = df.withColumn("result", max_udf(f_udf(df.col1), f_udf(df.col2))) 

Uwaga:

Drugie podejście jest ważna tylko wtedy, gdy funkcje wewnętrzne (tutaj f_udf) generowanie poprawnych wyrażeń SQL.

Działa tutaj ponieważ f_udf(df.col1) i f_udf(df.col2) są oceniane jako Column<b'(col1 + 1)'> i Column<b'(col2 + 1)'> odpowiednio, zanim przeszedł do max_udf. Nie działałby z funkcją dowolną.

To nie będzie działać, jeśli spróbujemy na przykład coś takiego:

from math import exp 

df.withColumn("result", max_udf(exp(df.col1), exp(df.col2)))