使用join时，Spark的迭代时间呈指数增长。

我对Spark还比较陌生，正在尝试使用迭代算法（期望最大化）来实现以马尔可夫模型表示的聚类。所以我需要进行迭代和连接操作。

我遇到的一个问题是，每次迭代的时间呈指数增长。

经过一些试验，我发现在进行迭代时，需要持久化要在下一次迭代中重用的RDD，否则每次迭代Spark都会创建一个执行计划，重新计算RDD，从而增加计算时间。

init = sc.parallelize(xrange(10000000), 3)
init.cache()
for i in range(6):
    print i
    start = datetime.datetime.now()
    init2 = init.map(lambda n: (n, n*3))        
    init = init2.map(lambda n: n[0])
#     init.cache()
    print init.count()    
    print str(datetime.datetime.now() - start)

结果为：

0
10000000
0:00:04.283652
1
10000000
0:00:05.998830
2
10000000
0:00:08.771984
3
10000000
0:00:11.399581
4
10000000
0:00:14.206069
5
10000000
0:00:16.856993

因此，添加cache()有所帮助，并且迭代时间变为恒定。

init = sc.parallelize(xrange(10000000), 3)
init.cache()
for i in range(6):
    print i
    start = datetime.datetime.now()
    init2 = init.map(lambda n: (n, n*3))        
    init = init2.map(lambda n: n[0])
    init.cache()
    print init.count()    
    print str(datetime.datetime.now() - start)
0
10000000
0:00:04.966835
1
10000000
0:00:04.609885
2
10000000
0:00:04.324358
3
10000000
0:00:04.248709
4
10000000
0:00:04.218724
5
10000000
0:00:04.223368

但是，在迭代中进行连接操作时，问题又出现了。

下面是一些简单的代码来演示这个问题。即使在每个RDD转换上进行缓存也无法解决问题：

init = sc.parallelize(xrange(10000), 3)
init.cache()
for i in range(6):
    print i
    start = datetime.datetime.now()
    init2 = init.map(lambda n: (n, n*3))
    init2.cache()
    init3 = init.map(lambda n: (n, n*2))
    init3.cache()
    init4 = init2.join(init3)
    init4.count()
    init4.cache()
    init = init4.map(lambda n: n[0])
    init.cache()
    print init.count()    
    print str(datetime.datetime.now() - start)

以下是输出结果。如您所见，迭代时间呈指数增长:(

0
10000
0:00:00.674115
1
10000
0:00:00.833377
2
10000
0:00:01.525314
3
10000
0:00:04.194715
4
10000
0:00:08.139040
5
10000
0:00:17.852815

真的非常感谢任何帮助 🙂