tf-idf 文档聚类与 Apache Spark 中的 K-means 将点放入一个聚类中

Question

我正在尝试通过预处理、生成 tf-idf 矩阵，然后应用 K-means 来完成对文本文档进行聚类的经典工作。但是，在经典的 20NewsGroup 数据集上测试此工作流程会导致大多数文档被聚集到一个集群中。 （我最初尝试将 20 个组中的 6 个中的所有文档聚类 - 因此希望聚类为 6 个聚类）。

我在 Apache Spark 中实现了这一点，因为我的目的是在数百万个文档中使用这种技术。这是在 Databricks 上用 Pyspark 编写的代码：

#declare path to folder containing 6 of 20 news group categories
path = "/mnt/%s/20news-bydate.tar/20new-bydate-train-lessFolders/*/*" % 
MOUNT_NAME

#read all the text files from the 6 folders. Each entity is an entire 
document. 
text_files = sc.wholeTextFiles(path).cache()

#convert rdd to dataframe
df = text_files.toDF(["filePath", "document"]).cache()

from pyspark.ml.feature import HashingTF, IDF, Tokenizer, CountVectorizer 

#tokenize the document text
tokenizer = Tokenizer(inputCol="document", outputCol="tokens")
tokenized = tokenizer.transform(df).cache()

from pyspark.ml.feature import StopWordsRemover

remover = StopWordsRemover(inputCol="tokens", 
outputCol="stopWordsRemovedTokens")
stopWordsRemoved_df = remover.transform(tokenized).cache()

hashingTF = HashingTF (inputCol="stopWordsRemovedTokens", outputCol="rawFeatures", numFeatures=200000)
tfVectors = hashingTF.transform(stopWordsRemoved_df).cache()    

idf = IDF(inputCol="rawFeatures", outputCol="features", minDocFreq=5)
idfModel = idf.fit(tfVectors)

tfIdfVectors = idfModel.transform(tfVectors).cache()

#note that I have also tried to use normalized data, but get the same result
from pyspark.ml.feature import Normalizer
from pyspark.ml.linalg import Vectors

normalizer = Normalizer(inputCol="features", outputCol="normFeatures")
l2NormData = normalizer.transform(tfIdfVectors)

from pyspark.ml.clustering import KMeans

# Trains a KMeans model.
kmeans = KMeans().setK(6).setMaxIter(20)
km_model = kmeans.fit(l2NormData)

clustersTable = km_model.transform(l2NormData)

ID number_of_documents_in_cluster
0    3024
3    5
1    3
5    2
2    2
4    1

正如您所见，我的大多数数据点都聚集到集群 0 中，我无法弄清楚我做错了什么，因为我在网上遇到的所有教程和代码都指向使用这种方法。

此外，我还尝试在 K-means 之前对 tf-idf 矩阵进行归一化，但这也会产生相同的结果。我知道余弦距离是一种更好的测量方法，但我希望在 Apache Spark 中使用标准 K-means 会提供有意义的结果。

谁能帮助我了解我的代码中是否存在错误，或者我的数据集群管道中是否缺少某些内容？

提前谢谢你！

这是python中的实现，即使具有大量最大特征，它也不会将所有文档组合在一起：

#imports
import pandas as pd
import os
import nltk
from nltk.tokenize import RegexpTokenizer
from nltk.corpus import stopwords

import numpy as np

from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.cluster import KMeans, MiniBatchKMeans 

vectorizer = TfidfVectorizer(max_features=200000, lowercase=True,
                             min_df=5, stop_words='english',
                             use_idf=True)

X = vectorizer.fit_transform(df['document'])

#Apply K-means to create cluster
from time import time

km = KMeans(n_clusters=20, init='k-means++', max_iter=20, n_init=1,
            verbose=False)

km.fit(X)

#result
3     2634
6     1720
18    1307
15     780
0      745
1      689
16     504
8      438
7      421
5      369
11     347
14     330
4      243
13     165
10     136
17     118
9      113
19     106
12      87
2       62

我原以为我们可以在 pyspark 中使用具有欧几里得距离的 KMeans 复制类似的东西，然后再在 KMeans 中尝试余弦或 Jaccard 距离。有什么解决方案或cmets？

Answer 1

@纳西尔， Spark k-means(scala mllib api) 在我的实验中也一直在产生高度倾斜的集群大小分布（见图 1）。大多数数据点分配给一个集群。该实验是使用 20 个具有基本事实的新闻组数据进行的：大约 10K 数据点被手动分类为相当平衡的 20 个组。 http://qwone.com/~jason/20Newsgroups/

最初我怀疑向量创建步骤（使用 Spark 的 HashingTF 和 IDF 库）是导致不正确聚类的原因。然而，即使在实现了我自己版本的基于 TF-IDF 的向量表示之后，我仍然得到了相似的聚类结果，但大小分布高度偏斜。

最终我在 spark 上实现了我自己的 k-means 版本，它使用标准 TF-IDF 矢量表示和 (-ve) 余弦相似度作为距离度量。这个 k-means 的结果看起来是正确的。请参见下图 2。

此外，我通过插入欧几里德距离作为相似性度量（与我自己的 kmean 版本）进行了实验，结果看起来仍然正确，不像 spark k-means 那样倾斜。

figure 1 and 2

Answer 2

只需几个快速的 cmets：

• K-Means 通常不是文本分析的最佳算法，因为它在高维度上表现不佳。我会推荐 LDA。
• 使用 K-Means，如果您将特征数量减少到大约 2000 个，那么您更有可能获得多个不同的集群。 （我在 Databricks CE 中 /databricks-datasets/news20.binary/data-001/training 提供的 20news 数据集上快速尝试了这个，并且能够获得不同的集群。）
• 不相关：如果将所有转换器和 K-Means 放入流水线，然后只调用一次 fit() 和 transform()，MLlib 代码可以不那么冗长。 :)

这是我从你修改的代码可以运行。警告：我根本没有调整它，所以集群目前非常无用（但它确实找到了不同的集群）。

df = spark.read.parquet("/databricks-datasets/news20.binary/data-001/training")
df.cache().count()

from pyspark.ml.feature import HashingTF, IDF, Tokenizer, CountVectorizer, StopWordsRemover
tokenizer = Tokenizer(inputCol="text", outputCol="tokens")
remover = StopWordsRemover(inputCol="tokens", outputCol="stopWordsRemovedTokens")
hashingTF = HashingTF(inputCol="stopWordsRemovedTokens", outputCol="rawFeatures", numFeatures=2000)
idf = IDF(inputCol="rawFeatures", outputCol="features", minDocFreq=5)

from pyspark.ml.clustering import KMeans
kmeans = KMeans(k=20)

from pyspark.ml import Pipeline
pipeline = Pipeline(stages=[tokenizer, remover, hashingTF, idf, kmeans])

model = pipeline.fit(df)

results = model.transform(df)
results.cache()

display(results.groupBy("prediction").count())  # Note "display" is for Databricks; use show() for OSS Apache Spark