查找每个 kmeans 集群的热门词

Question

我有以下代码部分将推文集合的 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 上看到，在 4 个具有 20.5 Gb 执行程序内存和 2 gb 驱动程序内存的 VM 集群上执行 reduceByKey() 操作时，我的代码需要大约 70 分钟。推文数量为 500K。文本文件大小为 31 Mb post 处理停用词和垃圾字符。

Answer 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)]
```
你只能开始一份工作，而不是每个集群的一份工作，而且开始一份工作并不便宜（例如，请参阅我对的回答）。在这里能收获多少取决于簇数
由于数据未被扁平化，因此可做的事情就更少了。串联列表不会给您带来任何好处，而且需要大量复制。使用字典可以减少存储的数据量，就地更新不需要副本。您可以尝试通过调整 merge_value:
来进一步改进
```
def merge_value(cnt, doc):
    for v in doc:
        cnt[v] += 1
    return cnt1
```

旁注：

有了 30 MB 的数据和 20.5 GB 的内存，我根本不会为 Spark 而烦恼。由于 k-means 只需要很少的额外内存，您可以在本地以更低的成本并行创建多个模型。

查找每个 kmeans 集群的热门词

Finding top words per kmeans cluster

python

apache-spark

pyspark

apache-spark-mllib