我目前正在实现这篇论文的实现，描述映射减少算法到连接组件：https://www.cse.unr.edu/~hkardes/pdfs/ccf.pdf 作为大数据世界的初学者，我开始使用一个小图实现 CCF-Iterate（w. 二次排序）算法：6 个边和 8 个节点。我正在使用免费版本的 Databricks 运行此代码。

给出结果需要1分钟。对于这样一个小例子来说，这似乎太长了。我怎样才能减少这个时间？什么样的优化是可能的？任何建议都会非常感激。 这个想法是测试这个算法的大图

PySpark 代码：

graph = sc.parallelize([ (2,3),(1,2),(2,4),(3,5),(6,7),(7,8)])
counter_new_pair = sc.accumulator(1)

while (counter_new_pair.value > 0):
  
  counter_new_pair = sc.accumulator(0)

  #CCF Iterate Sorting
  mapping_1 = graph.map(lambda x : (x[0],x[1]))
  mapping_2 = graph.map(lambda x : (x[1],x[0]))
  fusion = mapping_1.union(mapping_2)
  fusion = fusion.groupByKey().map(lambda x : (x[0],list(x[1])))

  fusion = fusion.map(lambda x : (x[0],sorted(x[1])))
  values = fusion.filter(lambda x : x[1][0] < x[0])
  
  key_min_value = values.map(lambda x : (x[0],x[1][0]))
  values = values.map(lambda x : (x[1][0],x[1][1:]))
  values = values.filter(lambda x : len(x[1]) != 0)
  values = values.flatMap(lambda x : [(val,x[0]) for val in x[1]])
  values.foreach(lambda x: counter_new_pair.add(1))
  joined = values.union(key_min_value)

  # CCF Dedup

  mapping = joined.map(lambda x : ((x[0],x[1]),None))
  graph = mapping.groupByKey().map(lambda x : (x[0][0],x[0][1]))
  
  

谢谢