如何降低Spark中approxSimilarityJoin引发的Shuffle Write？

Hey there! Let's tackle this Jaccard similarity scaling problem you're facing with Spark's approxSimilarityJoin—that 270GB shuffle write and 3+ hour runtime are definitely way out of line for a 16GB dataset. Here are actionable optimizations for code, cluster config, plus alternative approaches to speed things up:

Code-Level Optimizations #

• Tweak MinHashLSH & HashingTF Parameters
  • First, setNumHashTables(3): More hash tables boost precision but drastically increase shuffle data (since each table adds to the hash signature length). If your use case can tolerate a tiny drop in precision, try reducing this to 2—you’ll see a noticeable cut in shuffle size.
  • For HashingTF.setNumFeatures(1048576): This 2^20 feature space is overkill if your actual unique string count is much smaller (e.g., hundreds of thousands). Drop it to a smaller power of two like 2^18 (262144) to shrink feature vectors, reducing both memory usage and shuffle write.
• Eliminate Redundant Self-Join Pairs
  • Your current self-join generates duplicate pairs (A-B and B-A). Add a filter to keep only one direction:

    val dffilter = model.approxSimilarityJoin(featurizedData, featurizedData, 0.45)
                        .filter("datasetA.id < datasetB.id")
    

  • This cuts your candidate pairs in half, slashing computation and shuffle overhead.
• Preprocess to Reduce Data Volume
  • Deduplicate your values sets first: If raw data has duplicate strings in a set, run values = values.distinct() to shrink feature vectors.
  • Filter out tiny sets: Collections with 1-2 elements rarely produce meaningful Jaccard scores. Add a filter like dfA.filter(size(col("values")) > 2) to remove these upfront.

Cluster Configuration Tuning #

• Optimize Shuffle Settings
  • Adjust spark.sql.shuffle.partitions: The default 200 is too low for your 3x64-core cluster. Set it to 400-500 (2-3x total cores) to create evenly sized shuffle partitions, reducing IO overhead from too-small partitions.
  • Enable compression: Ensure spark.shuffle.compress and spark.shuffle.spill.compress are set to true (default in most versions) to compress shuffle data—this alone can cut shuffle write by 50%+.
  • Turn on Adaptive Query Execution (AQE): For Spark 3.x+, set spark.sql.adaptive.enabled = true. AQE automatically merges small partitions, optimizes shuffle sizes, and adjusts execution plans at runtime.
• Maximize Resource Utilization
  • Allocate executor resources properly: For each 64GB/64-core worker, set spark.executor.memory = 56G (leave 8GB for system processes) and spark.executor.cores = 64 (or split into 2 executors with 32 cores each if you prefer more parallelism).
  • Cache intermediate data: Cache featurizedData to avoid recomputing HashingTF results for the self-join:

    val featurizedData = hashingTF.transform(dfA).persist(StorageLevel.MEMORY_AND_DISK_SER)
    

    Using MEMORY_AND_DISK_SER ensures data is serialized, saving memory space.
• Use Columnar Storage
  • Convert your input data to Parquet/ORC format. These columnar formats compress better than raw text/CSV, reduce read IO, and let Spark process data more efficiently.

Alternative Approaches for Large-Scale Jaccard Calculation #

• Try SimHashLSH Instead of MinHashLSH
  • SimHash generates shorter hash signatures and is more efficient for high-dimensional text data. Swap out MinHash for SimHash:

    val sh = new SimHashLSH()
      .setNumHashTables(2)
      .setInputCol("features")
      .setOutputCol("hashes")
    val model = sh.fit(featurizedData)
    

  • You’ll likely see smaller shuffle sizes and faster runtime with minimal precision loss.
• Prefix Filtering for Exact Jaccard Scores
  • If you need exact (not approximate) results, implement prefix filtering: Sort each string set, take the first k elements as a prefix, and only compute Jaccard for pairs whose prefixes overlap. This reduces the number of candidate pairs drastically compared to a full cartesian join.
• Check for Data Skew
  • If a small number of hash buckets hold most of your data, it’s causing skew. Run featurizedData.groupBy("hashes").count().orderBy(desc("count")) to identify hot buckets. Adjust numFeatures in HashingTF or numHashTables in MinHash to redistribute data more evenly.

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