Pyspark - Display Top 10 words of document

user_8039968 0
Viewed 497

I'm quite new to Pyspark and did a tfidf processing on a dataframe with the following code

from pyspark.ml.feature import HashingTF, IDF, Tokenizer, CountVectorizer
from pyspark.sql.types import *
from pyspark.sql.functions import udf
wordsData = spark.createDataFrame([(0,"Hello","World","Spark","Is","Awesome","Hello","World"]),(1,["Hello","World","Spark","Is","Awesome","Hello","World"]),(2,["Hello","World"]),(3,["PYTHON", "Is", "Pretty", "Awesome"])],["label","words"])

#hashingTF way
hashingTF = HashingTF(inputCol="words", outputCol="rawFeatures")
featurizedData = hashingTF.transform(wordsData)
idf = IDF(inputCol="rawFeatures", outputCol="features")
idfModel = idf.fit(featurizedData)
rescaledData = idfModel.transform(featurizedData)
rescaledData.show(truncate=False)
print(hashingTF.indexOf("PYTHON"))

Now I want to store the Top 10 words with their tfidf value in a separate column. But since I'm not really used to working with vectors I'm a little stuck on how to achieve this. I know I somehow need to apply the indexOf function to every token of a document to find a mapping to its value, but I don't know how to do it. As far as I understood each vector is built up like this: (Size,[Key],[Value])

I was also thinking of using the CounteVectorizer way (and for that using its vocabulary) but I run in the same problem there.

Anyone who can help? This is the output so far:

+-----+------------------------------------------------+----------------------------------------------------------------+-----------------------------------------------------------------------------------------------------------------------------------------------+
|label|words                                           |rawFeatures                                                     |features                                                                                                                                       |
+-----+------------------------------------------------+----------------------------------------------------------------+-----------------------------------------------------------------------------------------------------------------------------------------------+
|0    |[Hello, World, Spark, Is, Awesome, Hello, World]|(262144,[32755,44691,64441,179674,262052],[2.0,2.0,1.0,1.0,1.0])|(262144,[32755,44691,64441,179674,262052],[0.44628710262841953,0.44628710262841953,0.22314355131420976,0.5108256237659907,0.22314355131420976])|
|1    |[Hello, World, Spark, Is, Awesome, Hello, World]|(262144,[32755,44691,64441,179674,262052],[2.0,2.0,1.0,1.0,1.0])|(262144,[32755,44691,64441,179674,262052],[0.44628710262841953,0.44628710262841953,0.22314355131420976,0.5108256237659907,0.22314355131420976])|
|2    |[Hello, World]                                  |(262144,[32755,44691],[1.0,1.0])                                |(262144,[32755,44691],[0.22314355131420976,0.22314355131420976])                                                                               |
|3    |[PYTHON, Is, Pretty, Awesome]                   |(262144,[61511,64441,191247,262052],[1.0,1.0,1.0,1.0])          |(262144,[61511,64441,191247,262052],[0.9162907318741551,0.22314355131420976,0.9162907318741551,0.22314355131420976])                           |
+-----+------------------------------------------------+----------------------------------------------------------------+-----------------------------------------------------------------------------------------------------------------------------------------------+
pyspark tf-idf nlp python
1 Answers

In your output data, rawFeatures and features are sparse vectors and it has 3 parts,size,indices,value.

for eg,(262144,[32755,44691,64441,179674,262052],[0.44628710262841953,0.44628710262841953,0.22314355131420976,0.5108256237659907,0.22314355131420976])

here size = 262144 ,indices = [32755,44691,64441,179674,262052] , values = [0.44628710262841953,0.44628710262841953,0.22314355131420976,0.5108256237659907,0.22314355131420976]

the indices are index values mapped to hash values of respective word. from https://spark.apache.org/docs/latest/mllib-feature-extraction.html#tf-idf
Our implementation of term frequency utilizes the hashing trick. A raw feature is mapped into an index (term) by applying a hash function. Then term frequencies are calculated based on the mapped indices.

Now to read output vector and map to word, we can use hash values of each word using same fitted model and map with features vector indices and get corresponding values.

1.first we get hashed index values for each word.

ndf = wordsData.select('label',f.explode('words').name('expwords')).withColumn('words',f.array('expwords'))
hashudf = f.udf(lambda vector : vector.indices.tolist()[0],StringType())
wordtf = hashingTF.transform(ndf).withColumn('wordhash',hashudf(f.col('rawFeatures')))
wordtf.show()
+-----+--------+---------+--------------------+--------+
|label|expwords|    words|         rawFeatures|wordhash|
+-----+--------+---------+--------------------+--------+
|    0|   Spark|  [Spark]|(262144,[179674],...|  179674|
|    0| Awesome|[Awesome]|(262144,[262052],...|  262052|
|    0|   World|  [World]|(262144,[32755],[...|   32755|
|    0|   Hello|  [Hello]|(262144,[44691],[...|   44691|
|    0|      Is|     [Is]|(262144,[64441],[...|   64441|
|    2|   Hello|  [Hello]|(262144,[44691],[...|   44691|
|    2|   World|  [World]|(262144,[32755],[...|   32755|
|    1|      Is|     [Is]|(262144,[64441],[...|   64441|
|    1|   World|  [World]|(262144,[32755],[...|   32755|
|    1| Awesome|[Awesome]|(262144,[262052],...|  262052|
|    1|   Spark|  [Spark]|(262144,[179674],...|  179674|
|    1|   Hello|  [Hello]|(262144,[44691],[...|   44691|
|    3| Awesome|[Awesome]|(262144,[262052],...|  262052|
|    3|  PYTHON| [PYTHON]|(262144,[191247],...|  191247|
|    3|  Pretty| [Pretty]|(262144,[61511],[...|   61511|
|    3|      Is|     [Is]|(262144,[64441],[...|   64441|
+-----+--------+---------+--------------------+--------+

2.flatten output features column to get indices & value.

udf1 = f.udf(lambda vec : dict(zip(vec.indices.tolist(),vec.values.tolist())),MapType(StringType(),StringType()))
valuedf = rescaledData.select('label',f.explode(udf1(f.col('features'))).name('wordhash','value'))
valuedf.show()
+-----+--------+-------------------+
|label|wordhash|              value|
+-----+--------+-------------------+
|    0|  179674| 0.5108256237659907|
|    0|   64441|0.22314355131420976|
|    0|   44691|0.44628710262841953|
|    0|   32755|0.44628710262841953|
|    0|  262052|0.22314355131420976|
|    1|  179674| 0.5108256237659907|
|    1|   64441|0.22314355131420976|
|    1|   44691|0.44628710262841953|
|    1|   32755|0.44628710262841953|
|    1|  262052|0.22314355131420976|
|    2|   44691|0.22314355131420976|
|    2|   32755|0.22314355131420976|
|    3|   64441|0.22314355131420976|
|    3|  191247| 0.9162907318741551|
|    3|  262052|0.22314355131420976|
|    3|   61511| 0.9162907318741551|
+-----+--------+-------------------+

3.get top n words for each document(label) filtering based on its rank and join both DFs and collect & sort to get the words along with its value.

w = Window.partitionBy('label').orderBy(f.desc('value'))
valuedf = valuedf.withColumn('rank',f.rank().over(w)).where(f.col('rank')<=3) # used 3 for testing.
valuedf.join(wordtf,['label','wordhash']).groupby('label').agg(f.sort_array(f.collect_list(f.struct(f.col('value'),f.col('expwords'))),asc=False).name('topn')).show(truncate=False)
+-----+-----------------------------------------------------------------------------------------------------------------------+
|label|topn                                                                                                                   |
+-----+-----------------------------------------------------------------------------------------------------------------------+
|0    |[{0.5108256237659907, Spark}, {0.44628710262841953, World}, {0.44628710262841953, Hello}]                              |
|1    |[{0.5108256237659907, Spark}, {0.44628710262841953, World}, {0.44628710262841953, Hello}]                              |
|3    |[{0.9162907318741551, Pretty}, {0.9162907318741551, PYTHON}, {0.22314355131420976, Is}, {0.22314355131420976, Awesome}]|
|2    |[{0.22314355131420976, World}, {0.22314355131420976, Hello}]                                                           |
+-----+-----------------------------------------------------------------------------------------------------------------------+
user_73583920
Related