python - 在并发.futures.ProcessPoolExecutor map() 和 Submit() 方法中使用 numpy.fromiter 和 numpy.array 的问题

Question

背景: 这个blog据报告，使用 numpy.fromiter() 相对于 numpy.array() 具有速度优势。使用提供的脚本作为基础，我想看看在 map() 和 submit() 中执行时 numpy.fromiter() 的好处python 的 concurrent.futures.ProcessPoolExecutor 类中的 code> 方法。

以下是我运行 2 秒的结果:

1. 很明显，当数组大小一般为 <256 时，numpy.fromiter() 比 numpy.array() 更快。​​
2. 但是，当由Python 的 concurrent.futures.ProcessPoolExecutor 类中的 map() 和 submit() 方法。

问题: 在 map() 和 submit 中使用时，numpy.fromiter() 和 numpy.array() 的性能是否会不一致且较差python 的 concurrent.futures.ProcessPoolExecutor 类中的 () 方法应该避免吗？如何改进我的脚本？

下面给出了我用于此基准测试的 Python 脚本。

map ():

#!/usr/bin/env python3.5
import concurrent.futures
from itertools import chain 
import time
import numpy as np
import pygal
from os import path

list_sizes = [2**x for x in range(1, 11)]
seconds = 2


def test(size_array):
    pyarray = [float(x) for x in range(size_array)]

    start = time.time()
    iterations = 0
    while time.time() - start <= seconds:
        np.fromiter(pyarray, dtype=np.float32, count=size_array)
        iterations += 1
    fromiter_count = iterations

    # array
    start = time.time()
    iterations = 0
    while time.time() - start <= seconds:
        np.array(pyarray, dtype=np.float32)
        iterations += 1
    array_count = iterations

    #return array_count, fromiter_count
    return size_array, array_count, fromiter_count


begin = time.time()
results = {}

with concurrent.futures.ProcessPoolExecutor(max_workers=6) as executor:
    data = list(chain.from_iterable(executor.map(test, list_sizes)))
    print('data = ', data)

for i in range( 0, len(data), 3 ):
    res = tuple(data[i+1:i+3])
    size_array = data[i]
    results[size_array] = res
    print("Result for size {} in {} seconds: {}".format(size_array,seconds,res))

out_folder = path.dirname(path.realpath(__file__))
print("Create diagrams in {}".format(out_folder))

chart = pygal.Line()
chart.title = "Performance in {} seconds".format(seconds)
chart.x_title = "Array size"
chart.y_title = "Iterations"

array_result = []
fromiter_result = []
x_axis = sorted(results.keys())
print(x_axis)
chart.x_labels = x_axis
chart.add('np.array', [results[x][0] for x in x_axis])
chart.add('np.fromiter', [results[x][1] for x in x_axis])
chart.render_to_png(path.join(out_folder, 'result_{}_concurrent_futures_map.png'.format(seconds)))

end = time.time()
compute_time = end - begin
print("Program Time = ", compute_time)

提交():

#!/usr/bin/env python3.5
import concurrent.futures
from itertools import chain 
import time
import numpy as np
import pygal
from os import path

list_sizes = [2**x for x in range(1, 11)]
seconds = 2


def test(size_array):
    pyarray = [float(x) for x in range(size_array)]

    start = time.time()
    iterations = 0
    while time.time() - start <= seconds:
        np.fromiter(pyarray, dtype=np.float32, count=size_array)
        iterations += 1
    fromiter_count = iterations

    # array
    start = time.time()
    iterations = 0
    while time.time() - start <= seconds:
        np.array(pyarray, dtype=np.float32)
        iterations += 1
    array_count = iterations

    return size_array, array_count, fromiter_count


begin = time.time()
results = {}

with concurrent.futures.ProcessPoolExecutor(max_workers=6) as executor:
    future_to_size_array = {executor.submit(test, size_array):size_array
                            for size_array in list_sizes}
    data = list(chain.from_iterable(
        f.result() for f in concurrent.futures.as_completed(future_to_size_array)))
    print('data = ', data)

for i in range( 0, len(data), 3 ):
    res = tuple(data[i+1:i+3])
    size_array = data[i]
    results[size_array] = res
    print("Result for size {} in {} seconds: {}".format(size_array,seconds,res))           

out_folder = path.dirname(path.realpath(__file__))
print("Create diagrams in {}".format(out_folder))

chart = pygal.Line()
chart.title = "Performance in {} seconds".format(seconds)
chart.x_title = "Array size"
chart.y_title = "Iterations"

x_axis = sorted(results.keys())
print(x_axis)
chart.x_labels = x_axis
chart.add('np.array', [results[x][0] for x in x_axis])
chart.add('np.fromiter', [results[x][1] for x in x_axis])
chart.render_to_png(path.join(out_folder, 'result_{}_concurrent_futures_submitv2.png'.format(seconds)))

end = time.time()
compute_time = end - begin
print("Program Time = ", compute_time)

序列号:(对 original code 进行细微更改)

#!/usr/bin/env python3.5
import time
import numpy as np
import pygal
from os import path

list_sizes = [2**x for x in range(1, 11)]
seconds = 2


def test(size_array):
    pyarray = [float(x) for x in range(size_array)]

    # fromiter
    start = time.time()
    iterations = 0
    while time.time() - start <= seconds:
        np.fromiter(pyarray, dtype=np.float32, count=size_array)
        iterations += 1
    fromiter_count = iterations

    # array
    start = time.time()
    iterations = 0
    while time.time() - start <= seconds:
        np.array(pyarray, dtype=np.float32)
        iterations += 1
    array_count = iterations

    return array_count, fromiter_count


begin = time.time()
results = {}

for size_array in list_sizes:
    res = test(size_array)
    results[size_array] = res
    print("Result for size {} in {} seconds: {}".format(size_array,seconds,res))

out_folder = path.dirname(path.realpath(__file__))
print("Create diagrams in {}".format(out_folder))

chart = pygal.Line()
chart.title = "Performance in {} seconds".format(seconds)
chart.x_title = "Array size"
chart.y_title = "Iterations"

x_axis = sorted(results.keys())
print(x_axis)
chart.x_labels = x_axis
chart.add('np.array', [results[x][0] for x in x_axis])
chart.add('np.fromiter', [results[x][1] for x in x_axis])
#chart.add('np.array', [x[0] for x in results.values()])
#chart.add('np.fromiter', [x[1] for x in results.values()])
chart.render_to_png(path.join(out_folder, 'result_{}_serial.png'.format(seconds)))

end = time.time()
compute_time = end - begin
print("Program Time = ", compute_time)

Answer 1

我之前遇到的 numpy.fromiter() 和 numpy.array() 性能不一致且较差的原因似乎与相关>concurrent.futures.ProcessPoolExecutor 使用的 CPU 数量。我之前使用过 6 个 CPU。下图显示了使用 2、4、6 和 8 个 CPU 时 numpy.fromiter() 和 numpy.array() 的相应性能。这些图表表明存在可以使用的最佳 CPU 数量。使用太多 CPU(即 >4 个 CPU)对于小数组(<512 个元素)可能是有害的。例如，与串行运行相比，>4 个 CPU 可能会导致性能降低(1/2 倍)，甚至性能不一致。