PySpark ligne sage fonction de la composition

Comme un exemple simplifié, j'ai un dataframe "df" avec des colonnes "col1,col2" et je veux calculer une ligne sage maximum après application d'une fonction à chaque colonne :

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)))

Donc si df:

col1   col2
1      2
3      0

Puis

df2:

col1   col2  result
1      2     3
3      0     4

Le ci-dessus ne semble pas fonctionner et produit "Impossible d'évaluer l'expression: PythonUDF#f..."

Je suis tout à fait positive "f_udf" fonctionne très bien sur ma table, et le principal problème est avec le max_udf.

Sans créer des colonnes supplémentaires ou de l'aide de la base map/reduce, est-il un moyen pour faire le dessus entièrement en utilisant dataframes et de l'udf? Comment dois-je modifier "max_udf"?

J'ai aussi essayé:

max_udf=udf(max, IntegerType())

qui produit la même erreur.

J'ai également confirmé que les travaux suivants:

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

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

Pourquoi est-ce que je peux pas faire en une seule fois?

Je voudrais voir une réponse qui généralise à toute la fonction "f_udf" et "max_udf."

InformationsquelleAutor Alex R. | 2016-04-12
  1. 33

    J'ai eu le même problème et trouvé la solution dans la réponse à la cette question stackoverflow

    Pour passer plusieurs colonnes ou une ligne entière à un UDF utiliser un 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()

    retourne:

    +----+----+----------+
|   a|   b|null_count|
+----+----+----------+
|null|null|         2|
|   1|null|         1|
|null|   2|         1|
+----+----+----------+
    • Merci, c'est le premier véritable réponse à cette question!
    • si vous êtes satisfait de cette réponse, veuillez accepter!
    InformationsquelleAutor Christoph Hösler
  2. 6

    UserDefinedFunction est jeter de l'erreur, tout en acceptant des fonctions définies par l'utilisateur comme leurs arguments.

    Vous pouvez modifier le max_udf comme ci-dessous pour le faire fonctionner. 

    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))

    Ou 

    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)))

    Note:

    La deuxième approche est valide si et seulement si les fonctions internes (ici f_udf) générer des expressions SQL valides.

    Cela fonctionne ici, car f_udf(df.col1) et f_udf(df.col2) sont évaluées comme Column<b'(col1 + 1)'> et Column<b'(col2 + 1)'> respectivement, avant d'être passé à max_udf. Il ne serait pas travailler avec une fonction arbitraire.

    Ça ne marcherait pas si nous essayons par exemple quelque chose comme ceci:

    from math import exp

df.withColumn("result", max_udf(exp(df.col1), exp(df.col2)))
    • Merci pour la réponse! Pourriez-vous clarifier la deuxième approche? Je ne comprends pas comment vous n'avez pas besoin f_udf être une bonne foi de l'UDF, de l'appliquer à la trame de données de la colonne?
    • La deuxième réponse semble être en exploitant le fait que dataframe colonnes de répondre aux "+" des opérations. Est-il quelque chose qui généralise à d'autres "f_udf"? En général, si j'ai un certain nombre de "f_udf"' fonctions, je dois écrire un ensemble distinct de max_udf fonctions pour chacun?
    • Je suis désolé, je suis également nouveau sur spark. J'ai remarqué que je peux faire des opérations sur les colonnes par des fonctions normales, sans les convertir en Udf. U peut pls l'élever comme une autre question? J'ai besoin de savoir le sna trop
    • Non, il n'y en a pas. En général, vous devriez préférer à l'aide de méthodes d'évaluation que les expressions SQL (comme les expressions arithmétiques) et l'utilisation de Python UDF seulement comme un dernier recours. Ils ont quelques très laid implications sur les performances et la portée relativement limitée. Si vous avez des complexes de fonctions Python vous souhaitez composer simplement le faire dans la plaine de Python et enveloppement haut niveau de l'appel avec l'UDF.
    InformationsquelleAutor Mohan