Pythonの並列処理で任意の順番＆形で結果を取得する

はじめに

最近Pythonで大量の演算処理を行うプログラムを書いた際，あまりにも計算に時間がかかるので並列処理を勉強し始めました．
そこで，自分の思い通りに使えるようにするにはどうすれば良いか考えていた所，とある形に落ち着いたので共有したいと思いました．

並列処理を使えるモジュールは，multiprocessingやconcurrentなどがありますが，この場ではconcurrentを用います．

(勉強が不十分でもっと良い書き方があるかもしれないので，もし左様でありましたら，ご教授いただけると幸いです．)

定番　コード①

まずPythonで並列処理を行いたい場合，大体このような形になると思います．
(本当はfor文でリスト内包表記を使用するべきなんですが，目を瞑っていただけると助かります．)

import time # 処理時間を測定するため
from concurrent import futures

ln = 40 # 適当な数字
data = list(range(ln))

def calculation(index): # 簡単な演算処理，普通に回すと大体17秒~18秒かかった．
    num = 0
    for i in range(10000000): num += 1
    return num, index

def main(): # メイン
    t0 = time.time()
    with futures.ProcessPoolExecutor() as executor:
        map_result = executor.map(calculation, data) # calculation()を40回分計算する
        result = list(map_result)
    t1 = time.time(); print(f'{t1-t0:.2f}秒\n')

if __name__ == '__main__':
    main()

これを実行すると計算の結果はこのようになります．

result = [(10000000, 0), (10000000, 1), ...省略... (10000000, 38), (10000000, 39)]

とてもシンプルで基本的にこれで十分だと思います．
しかし，個人的な意見として，今後 複雑な計算処理を行いたいとき，この出力結果の取得方法はとても分かりづらく，自由が効かないと感じました．

自分が求める要件はこんな感じです．

• 決まった順番，ランダムな順番ではなく，自分で定めた任意の順番で出力結果を得たい
• 自分の好きな形(numpyの行列など)で出力結果を格納出来るようにしたい
• 状況に応じて複数の値を入力，出力出来るようにしたい．
• 今後numbaを使う可能性を考えて，なるべく外部変数へのアクセスを避けたい．
• tqdmを用いてfor文の進捗を表示させるとき，自分の思う通りに簡単にカスタマイズ出来るようにしたい．

そこで色々いじってみた結果，最終的に以下のコードを簡単なテンプレートとして今後使いまわそうと考えました．

最終案　コード②

import time # 処理時間を測定するため
import numpy as np # 行列の演算では不可欠．
from concurrent import futures

ln = 40 # 適当な数字
data = list(range(ln))

def calculation(index): # 簡単な演算処理，普通に回すと大体17秒~18秒かかった．
    num = 0
    for i in range(10000000): num += 1
    return num, index

def main(): # メイン
    t0 = time.time()
    log = np.zeros((ln,2)) # 出力結果格納用
    with futures.ProcessPoolExecutor() as executor:
        for i in range(ln): # 計算開始
            # 変数(プロセス?)の名前を変えながら，演算処理のプロセスを開始，順番はほぼ関係ない．
            exec(f'result{i} = executor.submit(calculation,{i})')
        for i in range(ln): # 結果を待つ，
            # rangeの中身などを変えることで任意の順番で結果を受け取る．
            exec(f'log[{i},:] = result{i}.result()') # 任意の形で結果を格納．
            # 今回は40x2の行列に格納．
    t1 = time.time();print(f'{t1-t0:.2f}秒\n')

if __name__ == '__main__':
    main()

これを実行すると，計算の結果はこのようになります．先程よりは見やすく，整理しやすい結果になったと思います．

log = [[1.0e+07 0.0e+00]
      [1.0e+07 1.0e+00]
      [1.0e+07 2.0e+00]
      ...省略...
      [1.0e+07 3.7e+01]
      [1.0e+07 3.8e+01]
      [1.0e+07 3.9e+01]]

このような感じで作成すれば，今後どのような計算プログラムを組みたくても上手く対応できそうに感じました．

実行速度に影響はあったか？

コード②では，プロセスの開始や結果の取得にfor文を用いましたが，ほとんどの場合実行速度に影響はないと見てよさそうです．このfor文は演算処理自体には影響を与えていないはずです．

コード①とコード②を何回か実行してみたところ，寧ろコード②の方が安定して速かった印象を受けました．以下に処理時間をまとめました．(大体の目安で，しっかりとデータを取った訳では無いです)

• 並列処理を使わない場合：およそ17~18秒．
• 並列処理を使う場合：
  • コード①：約3.4 ~ 4秒．バラツキが大きかった．(原因は分からないです)
  • コード②：約3.3 ~ 3.5秒，バラツキが小さかった．

追記

コード②ではグローバル宣言しなくても動作しましたが，
基本execで変数を作成するときは，グローバル宣言をするか，第2引数にglobals()を渡さなくてはいけないです．
例：

exec('global a; a=1')
exec('a=1', globals())

実行環境

Windows 11，Ryzen 7 5800H，RAM16GB（DDR4）