查找每个 kmeans 集群的热门单词答案

【问题标题】：Finding top words per kmeans cluster查找每个 kmeans 集群的热门单词
【发布时间】：2015-11-08 21:02:07
【问题描述】：

我有以下代码部分，将推文集合的 TFIDF 映射到原始单词，然后用于查找每个集群中的热门单词：

#document = sc.textFile("<text file path>").map(lambda line: line.split(" "))
#"tfidf" is an rdd of tweets contained in "document"
#map tfidf to original tweets and cluster similar tweets
clusterIds = clusters.predict(tfidf)
mapped_value = clusterIds.zip(document)
cluster_value = mapped_value.reduceByKey(lambda a,b: a+b).take(cluster_num)


#Fetch the top 5 words from each cluster
topics = []
for i in cluster_value:
    word_count = sc.parallelize(i[1])
    topics.append(
        word_count.map(lambda x: (x,1))
            .reduceByKey(lambda x,y: x+y)
            .takeOrdered(5, key=lambda x: -x[1]))

有没有更好的方法来做到这一点？我在 Spark UI 上看到，在具有 20.5 Gb 执行程序内存和 2 Gb 驱动程序内存的 4 个 VM 的集群上执行 reduceByKey() 操作时，我的代码需要大约 70 分钟。推文数量为 500K。文本文件大小为 31 Mb 后处理停用词和垃圾字符。

【问题讨论】：

标签： python apache-spark pyspark apache-spark-mllib

【解决方案1】：

由于您没有提供a Minimal, Complete, and Verifiable example，我只能假设document rdd 包含标记化文本。因此，让我们创建一个虚拟示例：

mapped_value = sc.parallelize(
    [(1, "aabbc"), (1, "bab"), (2, "aacc"), (2, "acdd")]).mapValues(list)
mapped_value.first()
## (1, ['a', 'a', 'b', 'b', 'c'])

您可以做的一件事是同时聚合所有集群：

from collections import Counter

create_combiner = Counter

def merge_value(cnt, doc):
    cnt.update(Counter(doc))
    return cnt

def merge_combiners(cnt1, cnt2):
    cnt1.update(cnt2)
    return cnt1

topics = (mapped_value
    .combineByKey(create_combiner, merge_value, merge_combiners)
    .mapValues(lambda cnt: cnt.most_common(2)))

topics
## [(1, [('b', 4), ('a', 3)]), (2, [('a', 3), ('c', 3)])]

您可以通过将 Counter 替换为普通的 dict 并手动计数/更新来进一步改进，但我认为这不值得大惊小怪。

有什么收获？

首先，您减少了必须移动的数据量（序列化 - 传输 - 反序列化）。特别是，您收集的不仅仅是为了将数据发送回工作人员。

收集和发送费用很高，因此除非这是唯一的选择，否则您应该避免这样做。如果对整个数据集的聚合过于昂贵，那么一种更可取的方法可能是重复 filter，相当于如下所示：
```
[rdd.filter(lambda (k, v): k == i).map(...).reduce(...)
    for i in range(number_of_clusters)]
```
您只开始一项工作，而不是每个集群的一项工作，并且开始一项工作并不便宜（例如，请参阅我对Spark MLLib's LassoWithSGD doesn't scale? 的回答）。您在这里可以获得多少取决于集群的数量。
由于数据没有被展平，所以要做的事情就更少了。连接列表什么都没有，而且需要大量的复制。使用字典可以减少存储的数据量，就地更新不需要副本。您可以通过调整merge_value来尝试进一步改进：
```
def merge_value(cnt, doc):
    for v in doc:
        cnt[v] += 1
    return cnt1
```

旁注：

拥有 30 MB 的数据和 20.5 GB 的内存，我根本不会使用 Spark。由于 k-means 需要的额外内存非常少，因此您可以以更低的成本在本地并行创建多个模型。

【讨论】：