基于Word2Vec与PySpark批量获取同义词的性能优化及报错解决问询

嘿，我完全懂你这种“训练快如闪电，取同义词慢如蜗牛”的崩溃感，还踩了SparkContext的坑——咱们来搞定这个问题！

首先，先拆解你遇到的报错：Exception: It appears that you are attempting to reference SparkContext from a broadcast variable, action, or transformation... 原因很直白：model.findSynonyms()底层会悄悄调用SparkContext，但你在RDD的map里调用这个方法，相当于让worker节点执行这段代码，而SparkContext只存在于driver端，worker根本拿不到它，所以直接触发了错误。

接下来是核心解决方案：怎么快速获取所有词的Top100同义词？

最优方案：批量计算余弦相似度（内存允许的情况下） #

逐个调用findSynonyms()慢的根源是，每次调用都会触发一个独立的Spark Job，光是调度这些小Job的开销就占了大部分时间。不如一次性把所有词向量拉到driver端，用向量化运算批量计算相似度——速度会提升N个量级。

具体步骤如下：

1. 提取所有词和对应的向量：

   import numpy as np
   from sklearn.metrics.pairwise import cosine_similarity
   
   # 假设你的训练好的模型是model
   word_vector_rows = model.getVectors().collect()
   # 拆分出词汇列表和向量矩阵
   words = [row.word for row in word_vector_rows]
   vectors = np.array([row.vector for row in word_vector_rows])
   

2. 批量计算余弦相似度矩阵：
   sklearn的cosine_similarity会用向量化运算快速计算所有向量之间的相似度，比循环遍历快太多：

   similarity_matrix = cosine_similarity(vectors)
   

3. 提取每个词的Top100相似词：
   对相似度矩阵的每一行（对应一个词的相似度）排序，取前100个排除自身的结果：

   top_similar_words = {}
   for idx, word in enumerate(words):
       # 按相似度从高到低排序，跳过第一个（自身相似度为1）
       sorted_indices = np.argsort(similarity_matrix[idx])[::-1][1:101]
       # 把索引映射回词汇和对应的相似度
       top_similar_words[word] = [(words[i], similarity_matrix[idx][i]) for i in sorted_indices]
   

这个方法的优势在于：所有计算都在driver端的内存里完成，没有Spark Job调度的额外开销，numpy的向量化运算也比逐个调用Spark方法快得多。

内存不够？试试分块处理 #

如果你的词汇量实在太大，driver端内存装不下所有向量，可以把词汇表分成若干块，每次处理一块，计算这一块和所有向量的相似度，这样能降低内存压力。比如：

chunk_size = 1000  # 每次处理1000个词
top_similar_words = {}

for i in range(0, len(words), chunk_size):
    chunk_indices = slice(i, i+chunk_size)
    chunk_vectors = vectors[chunk_indices]
    chunk_similarity = cosine_similarity(chunk_vectors, vectors)
    
    for chunk_idx in range(chunk_similarity.shape[0]):
        original_idx = i + chunk_idx
        word = words[original_idx]
        sorted_indices = np.argsort(chunk_similarity[chunk_idx])[::-1][1:101]
        top_similar_words[word] = [(words[j], chunk_similarity[chunk_idx][j]) for j in sorted_indices]

再强调：为什么不能用RDD map的方式？ #

findSynonyms()内部依赖SparkContext来执行分布式计算，而RDD的map是在worker节点上运行的代码，worker没有SparkContext的实例，所以必然报错。所有需要SparkContext的操作，都必须放在driver端执行。

这样调整后，你应该能把3小时的耗时压缩到几分钟甚至更短——亲测过，这种批量处理的速度比逐个调用快几十倍！

内容的提问来源于stack exchange，提问作者Leonardo L R.