Le passage d'un bloc de données de la colonne et de la liste externe à l'udf en vertu de withColumn

J'ai une Étincelle dataframe avec la structure suivante. Le bodyText_token a les jetons (traité/jeu de mots). Et j'ai une liste imbriquée de mots-clés définis

root
 |-- id: string (nullable = true)
 |-- body: string (nullable = true)
 |-- bodyText_token: array (nullable = true)

keyword_list=['union','workers','strike','pay','rally','free','immigration',],
['farmer','plants','fruits','workers'],['outside','field','party','clothes','fashions']]

J'avais besoin de vérifier le nombre de jetons d'entrer dans chaque liste de mots clés et ajouter le résultat dans une nouvelle colonne de l'existant dataframe.
Par exemple: si tokens =["become", "farmer","rally","workers","student"]
le résultat sera -> [1,2,0]

La fonction suivante a fonctionné comme prévu.

def label_maker_topic(tokens,topic_words):
    twt_list = []
    for i in range(0, len(topic_words)):
        count = 0
        #print(topic_words[i])
        for tkn in tokens:
            if tkn in topic_words[i]:
                count += 1
        twt_list.append(count)

    return twt_list

J'ai utilisé de l'udf, en vertu de withColumn pour accéder à la fonction et j'obtiens une erreur. Je pense que c'est sur la transmission d'une liste externe à l'udf. Est il possible que je peux passer de la liste externe et la datafram colonne à une udf et ajouter une nouvelle colonne à mon dataframe?

topicWord = udf(label_maker_topic,StringType())
myDF=myDF.withColumn("topic_word_count",topicWord(myDF.bodyText_token,keyword_list))

OriginalL'auteur Jay | 2016-05-24

  1. 27

    La solution la plus propre est de passer des arguments supplémentaires à l'aide de la fermeture: 

    def make_topic_word(topic_words):
     return udf(lambda c: label_maker_topic(c, topic_words))

df = sc.parallelize([(["union"], )]).toDF(["tokens"])

(df.withColumn("topics", make_topic_word(keyword_list)(col("tokens")))
    .show())

    Cela ne nécessite pas de modifications dans keyword_list ou la fonction que vous envelopper avec l'UDF. Vous pouvez également utiliser cette méthode pour passer d'un objet arbitraire. Cela peut être utilisé pour passer par exemple d'une liste de sets pour l'efficacité des recherches.

    Si vous souhaitez utiliser votre actuelle UDF et passer topic_words directement, vous devrez le convertir en une colonne littérale d'abord:

    from pyspark.sql.functions import array, lit

ks_lit = array(*[array(*[lit(k) for k in ks]) for ks in keyword_list])
df.withColumn("ad", topicWord(col("tokens"), ks_lit)).show()

    En fonction de vos données et de vos exigences, il peut alternatifs, des solutions plus efficaces, qui ne nécessitent pas l'Udf (exploser + agrégat + effondrement) ou de recherche (hachage + vecteur des opérations).

    Je pense que vous devez aussi préciser le type de votre udf appel dans make_topic_word

    OriginalL'auteur zero323

  2. 8

    Le suivant fonctionne très bien où tous les paramètres externes peuvent être transmises à l'UDF a modifié le code pour aider quelqu'un) 

    topicWord=udf(lambda tkn: label_maker_topic(tkn,topic_words),StringType())
myDF=myDF.withColumn("topic_word_count",topicWord(myDF.bodyText_token))
    Cela fonctionne, mais je serais prudent avec cela, parce que l'udf aura la topic_words de la valeur au moment de l'udf a été défini. La modification du topic_words et de ré-utilisation de l'udf, plus tard, ne fonctionne pas - il va utiliser la valeur de topic_words au moment de l'udf a été défini.

    OriginalL'auteur Jay