使用 pyspark 并行化排序算法

Question

早上好，我开发了一个简单的合并排序算法，我想比较它在并行化和未并行化时的性能。

首先，我正在生成一个数字列表进行排序，并检查合并排序对列表进行排序需要多长时间。

接下来我想做的是将数字列表传递到sc.parallelize()并将list转换为RDD然后将合并排序函数传递到mapPartitions()然后collect()。

import random
import time
from pyspark import SparkContext
def execute_merge_sort(generated_list):
start_time = time.time()
sorted_list = merge_sort(generated_list)
elapsed = time.time() - start_time
print('Simple merge sort: %f sec' % elapsed)
return sorted_list

def generate_list(length):
N = length
generated_list = [random.random() for num in range(N)]
return generated_list
def merging(left_side, right_side):
result = []
i = j = 0
while i < len(left_side) and j < len(right_side):
if left_side[i] <= right_side[j]:
result.append(left_side[i])
i += 1
else:
result.append(right_side[j])
j += 1
if i == len(left_side):
result.extend(right_side[j:])
else:
result.extend(left_side[i:])
return result

def merge_sort(generated_list):
if len(generated_list) <= 1:
return generated_list
middle_value = len(generated_list) // 2
sorted_list = merging(merge_sort(generated_list[:middle_value]), merge_sort(generated_list[middle_value:]))
return sorted_list

def is_sorted(num_array):
for i in range(1, len(num_array)):
if num_array[i] < num_array[i - 1]:
return False
return True
generate_list = generate_list(500000)
sorted_list = execute_merge_sort(generate_list)
sc = SparkContext()
rdd = sc.parallelize(generate_list).mapPartitions(execute_merge_sort).collect()

当我执行此sc.parallelize(generate_list).mapPartitions(execute_merge_sort).collect()时，出现以下错误：

File "<ipython-input-15-1b7974b4fa56>", line 7, in execute_merge_sort
File "<ipython-input-15-1b7974b4fa56>", line 36, in merge_sort
TypeError: object of type 'itertools.chain' has no len()

任何帮助将不胜感激。提前谢谢。

Answer 1

我想出了如何解决TypeError: 'float' object is not iterable的问题。

这可以通过使用flatMap(lambda x: x)平展数据并调用glom()来解决，以便包装列表并使其可由函数execute_merge_sort执行。 通过执行以下行，返回的结果是一个包含排序列表的列表。

sc.parallelize(random_list_of_lists).flatMap(lambda x: x).glom().mapPartitions(execute_merge_sort_rdd).collect()