Die übergabe eines Daten-frame-Spalte und eine externe Liste mit udf unter withColumn

Ich habe eine Spark-dataframe folgende Struktur. Die bodyText_token hat das Token (bearbeitet/eingestellt von Wörtern). Und ich habe eine verschachtelte Liste mit den definierten keywords

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']]

Ich benötigt, um zu überprüfen, wie viele Münzen fallen unter den einzelnen keyword-Liste und fügen Sie das Ergebnis in einer neuen Spalte der bestehenden dataframe.
ZB: wenn tokens =["become", "farmer","rally","workers","student"]
das Ergebnis -> [1,2,0]

Folgende Funktion funktioniert wie erwartet.

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

Ich verwendete udf-unter withColumn, um die Funktion aufzurufen und ich bekomme eine Fehlermeldung. Ich denke, es ist zum übergeben einer externen Liste auf einer udf. Gibt es eine Möglichkeit, die ich übergeben können eine externe Liste, und die datafram Spalte zu einer udf-Datei und fügen Sie eine neue Spalte in meine dataframe?

topicWord = udf(label_maker_topic,StringType())
myDF=myDF.withColumn("topic_word_count",topicWord(myDF.bodyText_token,keyword_list))
InformationsquelleAutor Jay | 2016-05-24
  1. 29

    Die sauberste Lösung ist die übergabe von zusätzlichen Argumenten mit Verschluss: 

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

    Diese erfordert keine änderungen in keyword_list oder die Funktion wickeln Sie mit UDF. Sie können diese Methode auch verwenden, um eine beliebige Objekt. Dies kann verwendet werden, um pass zum Beispiel eine Liste von sets für effiziente Suchvorgänge.

    Wenn Sie wollen verwenden Sie Ihren aktuellen UDF und pass topic_words direkt haben Sie es zu konvertieren, um eine Spalte Literale zuerst:

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

    Abhängig von Ihren Daten und Anforderungen kann es alternative, effizientere Lösungen, die nicht verlangen, UDFs (explodieren + Aggregat + Kollaps) oder lookups (hashing + Vektor-Operationen).

    • Ich denke, Sie müssen auch den Typ angeben, in Ihrem udf Anruf in make_topic_word
    InformationsquelleAutor zero323
  2. 8

    Folgende Werke gut, wo alle externen parameter übergeben werden kann, um die UDF (eine optimierte code, um jemandem zu helfen) 

    topicWord=udf(lambda tkn: label_maker_topic(tkn,topic_words),StringType())
myDF=myDF.withColumn("topic_word_count",topicWord(myDF.bodyText_token))
    • Dies funktioniert, aber ich wäre vorsichtig mit diesem, weil die udf wird die topic_words Wert im moment der udf definiert wurde. So ändern Sie die topic_words - und re-mit dem udf-später wird nicht funktionieren - es wird weiterhin der Wert von topic_words an der Zeit die udf definiert wurde.
    InformationsquelleAutor Jay
Schreibe einen Kommentar