Aggregat sparse Vektor in pyspark

Question

Ich habe eine Hive-Tabelle, die Textdaten und einige Metadaten für jedes Dokument enthält. Sieht aus wie das.

from pyspark.ml.feature import Tokenizer 
from pyspark.ml.feature import CountVectorizer 

df = sc.parallelize([ 
    ("1", "doc_1", "fruit is good for you"), 
    ("2", "doc_2", "you should eat fruit and veggies"), 
    ("2", "doc_3", "kids eat fruit but not veggies") 
]).toDF(["month","doc_id", "text"]) 


+-----+------+--------------------+ 
|month|doc_id|    text| 
+-----+------+--------------------+ 
| 1| doc_1|fruit is good for...| 
| 2| doc_2|you should eat fr...| 
| 2| doc_3|kids eat fruit bu...| 
+-----+------+--------------------+ 

Ich möchte Wörter nach Monat zählen. Bisher habe ich einen CountVectorizer Ansatz:

tokenizer = Tokenizer().setInputCol("text").setOutputCol("words") 
tokenized = tokenizer.transform(df) 

cvModel = CountVectorizer().setInputCol("words").setOutputCol("features").fit(tokenized) 
counted = cvModel.transform(tokenized) 

+-----+------+--------------------+--------------------+--------------------+ 
|month|doc_id|    text|    words|   features| 
+-----+------+--------------------+--------------------+--------------------+ 
| 1| doc_1|fruit is good for...|[fruit, is, good,...|(12,[0,3,4,7,8],[...| 
| 2| doc_2|you should eat fr...|[you, should, eat...|(12,[0,1,2,3,9,11...| 
| 2| doc_3|kids eat fruit bu...|[kids, eat, fruit...|(12,[0,1,2,5,6,10...| 
+-----+------+--------------------+--------------------+--------------------+ 

Jetzt habe ich von Monat gruppieren möchten und etwas zurückgeben, die wie folgt aussieht: Wie

month word count 
1  fruit 1 
1  is  1 
... 
2  fruit 2 
2  kids 1 
2  eat 2 
... 

könnte ich das tun?

Answer 1

Es gibt keinen integrierten Mechanismus für Vector Aggregation, aber Sie brauchen hier keinen. Nachdem Sie die Daten in Token aufgeteilt haben, können Sie nur explode und Aggregat:

from pyspark.sql.functions import explode 

(counted 
    .select("month", explode("words").alias("word")) 
    .groupBy("month", "word") 
    .count()) 

Wenn Sie die Ergebnisse der vocabulary zu begrenzen bevorzugen fügen Sie einfach einen Filter:

from pyspark.sql.functions import col 

(counted 
    .select("month", explode("words").alias("word")) 
    .where(col("word").isin(cvModel.vocabulary)) 
    .groupBy("month", "word") 
    .count())