Spark Scala中计算余弦相似度后top_sim_map为何为空？

Let's break down why your top_sim_map ends up empty, and how to fix it properly:

Core Reasons for Empty Map #

1. Immutable Map + Distributed Execution Isolation #

• You defined top_sim_map as an immutable val Map in the Driver. When you try to do top_sim_map += {moid->simArr} inside the RDD foreach method, this operation creates a new Map instance—but this new Map only exists in the Executor's task process, not in the Driver's memory.
• Spark's RDD operations (like foreach) run on cluster Executors, which are separate JVM processes from the Driver. Any modifications to Driver-side variables inside Executor code won't sync back to the Driver—their memory spaces are completely isolated.

2. Inefficient (and Risky) Local Collection #

On top of the empty map issue, collecting the entire RDD to the Driver with collect() then broadcasting it is a bad practice for large datasets: it can crash the Driver with out-of-memory errors, and the O(n²) pairwise comparison will become unbearably slow as your data grows.

Correct Implementation #

Instead of trying to modify a Driver-side Map from Executors, use Spark's distributed operations to compute the results, then collect the final output to the Driver as a Map.

Step 1: Optimized Cosine Similarity Function #

First, optimize the similarity calculation to leverage SparseVector's structure (avoid iterating all dimensions):

import org.apache.spark.ml.linalg.SparseVector

def cosineSimilarity(vectorA: SparseVector, vectorB: SparseVector): Double = {
  val dotProduct = vectorA.values.zip(vectorB.values).map { case (a, b) => a * b }.sum
  val normA = Math.sqrt(vectorA.values.map(x => x * x).sum)
  val normB = Math.sqrt(vectorB.values.map(x => x * x).sum)
  
  // Handle zero-norm edge case to avoid division by zero
  if (normA == 0.0 || normB == 0.0) 0.0 else dotProduct / (normA * normB)
}

Step 2: Distributed Pairwise Similarity & Top10 Calculation #

Use RDD operations to compute pairwise similarities, group by each moid, and extract the top 10 similar entries:

import org.apache.spark.sql.Row

// Generate all pairwise combinations (exclude self-comparison)
val pairwiseRDD = dataframe.rdd.cartesian(dataframe.rdd)
  .filter { case (row1, row2) => row1.getString(0) != row2.getString(0) }
  .map { case (row1, row2) =>
    val moid1 = row1.getString(0)
    val moid2 = row2.getString(0)
    val similarity = cosineSimilarity(row1.getAs[SparseVector](1), row2.getAs[SparseVector](1))
    (moid1, (moid2, similarity))
  }

// Group by each moid, sort by similarity descending, take top 10
val top10RDD = pairwiseRDD.groupByKey()
  .mapValues { similarEntries =>
    similarEntries.toArray.sortBy(-_._2).take(10)
  }

// Collect results to Driver and convert to Map
val top_sim_map: Map[String, Array[(String, Double)]] = top10RDD.collect().toMap

Step 3: Verify the Result #

Now when you check top_sim_map.size or iterate over it, you'll see the expected data instead of an empty Map.

Key Improvements #

• No Driver-Executor Variable Sync Issues: All computation happens in distributed RDD operations, with results collected back to the Driver only at the end.
• Better Performance: The optimized cosine similarity avoids unnecessary iterations, and grouping/sorting happens on Executors instead of the Driver.
• Scalability Note: For large datasets, replace the Cartesian product with an approximate similarity algorithm like Locality Sensitive Hashing (LSH) to avoid O(n²) complexity.

内容的提问来源于stack exchange，提问作者zeffer