PySpark执行minhash相似度过滤速度极慢的优化咨询

分区调整建议 #

你不应该减少分区数量，当前4个分区属于过少的情况，会导致集群CPU资源无法充分利用：如果你的可用计算核心数大于4，绝大多数核心会处于闲置状态，所有计算压力都集中在4个核心串行执行。
建议将分区数调整为可用CPU核心数的1_{2倍，比如可用8核时调整为8}16个分区，可通过如下代码实现：

# 重分区优化并行度
df = df.repartition(16)

注意不要设置过高的分区数，否则会带来额外的任务调度开销，反而降低执行效率。

其他降低耗时的优化方案 #

• 修正过滤列笔误
  你示例代码中的过滤条件写的是df.filter(df.column > .5)，实际应该过滤minhash_sim列，若实际运行时写错了列名，会导致过滤逻辑失效，甚至触发全量扫描无意义数据。
• 优化Jaccard相似度计算逻辑
  你当前的实现需要同时计算数组交集和并集的长度，实际上Jaccard相似度可通过公式 |A∩B|/(|A| + |B| - |A∩B|) 推导，提前预存每个minhash_array的长度即可省去array_union的计算开销，修改后逻辑如下：

  # 提前预存minhash数组长度，避免重复计算
  df = df.withColumn('minhash_arr_len', size('minhash_array'))
  target_sig_len = lit(len(minhash_sig))
  intersect_cnt = size(array_intersect(minhash_array0, 'minhash_array'))
  df = df.withColumn('minhash_sim', intersect_cnt / (target_sig_len + col('minhash_arr_len') - intersect_cnt))
  

• 广播常量签名数组
  你用于计算相似度的目标minhash签名是全局常量，可通过广播变量避免每个task重复生成该数组，减少冗余计算：

  from pyspark.sql.functions import array, lit
  
  # 广播常量签名
  sig_bc = spark.sparkContext.broadcast([str(x) for x in minhash.signature(doc)])
  minhash_array0 = array([lit(i) for i in sig_bc.value])
  

• 添加前置过滤条件
  过滤前的df.head()仅需要读取第一个分区的第一条数据即可返回，所以速度极快；添加过滤条件后，Spark需要扫描所有分区数据直到找到符合条件的记录，若符合相似度阈值的记录占比极低，就需要扫描全量数据。你可以添加前置过滤规则提前筛掉不可能符合条件的记录，比如：

  # 长度差过大的数组Jaccard相似度不可能达到0.5，提前过滤
  df = df.filter(abs(col('minhash_arr_len') - len(minhash_sig)) <= len(minhash_sig))
  

• 使用MinHashLSH加速过滤
  若你需要多次执行相似度查询，可直接使用Spark MLlib内置的MinHashLSH实现提前对数据做分桶索引，后续相似度查询仅需要扫描对应分桶的数据，不需要全量计算，性能可提升数倍。
• 检查数据倾斜
  你可以通过如下代码检查4个分区的数据量是否均匀，若存在个别分区数据量远高于其他分区的倾斜问题，重分区即可解决长尾耗时：

  # 查看每个分区的行数
  print(df.rdd.mapPartitions(lambda it: [sum(1 for _ in it)]).collect())
  

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