python - 查找每个 kmeans 集群的 HitTest 门单词

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

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)])]

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

有什么收获？

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

  收集和发送的费用很高，因此您应该避免这样做，除非这是唯一的选择。如果整个数据集的聚合成本高昂，更好的方法可能是重复的过滤器，相当于这样:

  [rdd.filter(lambda (k, v): k == i).map(...).reduce(...)
      for i in range(number_of_clusters)]
  

• 您只启动一项工作，而不是每个集群启动一项工作，并且启动一项工作并不便宜(例如，请参阅我对 Spark MLLib's LassoWithSGD doesn't scale? 的回答)。您在这里可以获得多少 yield 取决于集群的数量。

• 由于数据没有被扁平化，所以要做的事情就更少了。连接列表不会给您带来任何好处，并且需要大量复制。使用字典可以减少存储的数据量，就地更新不需要副本。您可以尝试通过调整 merge_value 来进一步改进:

  def merge_value(cnt, doc):
      for v in doc:
          cnt[v] += 1
      return cnt1
  

<小时/>

旁注:

• 如果有 30 MB 的数据和 20.5 GB 的内存，我根本不会使用 Spark。由于 k 均值需要很少的额外内存，因此您可以以低得多的成本在本地并行创建多个模型。