Python ProcessPoolExecutorとは？CPU処理を並列化する方法
PythonのProcessPoolExecutorは、複数のプロセスを使って計算処理を並列化するクラスです。
GILの制約を受けないため、CPUを使い切る処理を本当に速くできます。
ProcessPoolExecutorの基本的な使い方
import concurrent.futures as cf
def heavy(n):
return sum(i * i for i in range(n))
if __name__ == "__main__":
with cf.ProcessPoolExecutor(max_workers=4) as ex:
print(list(ex.map(heavy, [100, 200, 300])))書き方はThreadPoolExecutorとまったく同じです。クラス名を変えるだけで切り替えられます。
ただしif __name__ == "__main__":が必須です。理由は後述します。
スレッドとの速度差
import concurrent.futures as cf, time, os
def heavy(n):
return sum(i * i for i in range(n))
if __name__ == "__main__":
N = 2_000_000
t = time.perf_counter()
with cf.ThreadPoolExecutor(max_workers=4) as ex:
list(ex.map(heavy, [N] * 4))
th = time.perf_counter() - t
t = time.perf_counter()
with cf.ProcessPoolExecutor(max_workers=4) as ex:
list(ex.map(heavy, [N] * 4))
pr = time.perf_counter() - t
print(f"Thread : {th:.2f}秒")
print(f"Process: {pr:.2f}秒 → {th/pr:.1f}倍")
print("CPU数:", os.cpu_count())実行結果（CPU 8コアの環境）：
Thread : 0.30秒
Process: 0.15秒 → 1.9倍
CPU数: 8スレッドでは速くなりません。GILにより、同時にPythonのコードを実行できるスレッドは1つだけだからです。
プロセスはそれぞれ独立したインタプリタを持つため、この制約を受けません。
I/O処理では逆に遅くなる
import concurrent.futures as cf, time
def wait(n):
time.sleep (0.3)
return n
if __name__ == "__main__":
for name, Ex in [("Thread", cf.ThreadPoolExecutor),
("Process", cf.ProcessPoolExecutor)]:
t = time.perf_counter()
with Ex(max_workers=4) as ex:
list(ex.map(wait, range(4)))
print(f"{name:8}: {time.perf_counter() - t:.2f}秒")実行結果：
Thread : 0.31秒
Process : 0.41秒プロセスのほうが遅くなりました。起動とデータの受け渡しにコストがかかるためです。
待ち時間が主体の処理にProcessを使わないでください。
__main__ガードが必須な理由
import concurrent.futures as cf
def f(n):
return n * 2
with cf.ProcessPoolExecutor(max_workers=2) as ex:
print(list(ex.map(f, [1, 2, 3])))実行結果（macOS・Windows）：
RuntimeError:
An attempt has been made to start a new process before the
current process has finished its bootstrapping phase.macOSとWindowsでは、子プロセスが親のスクリプトを最初から読み込み直します。ガードがないと、読み込むたびにまた新しいプロセスを作ろうとし、無限に増殖します。
if __name__ == "__main__":の中は子プロセスでは実行されないため、これで止まります。
Linuxでは別の方式（fork）が既定のため動いてしまうことがありますが、環境を選ばず必ず書いてください。
同じ理由で、Jupyter NotebookではProcessPoolExecutorが動かないことがあります。処理を別の.pyファイルに切り出してimportすれば回避できます。
渡すデータはpickle化できる必要がある
プロセス間でやり取りするため、関数も引数も戻り値もpickle化されます。
import concurrent.futures as cf
if __name__ == "__main__":
with cf.ProcessPoolExecutor() as ex:
list(ex.map(lambda x: x * 2, [1, 2, 3]))実行結果：
... Can't pickle <function <lambda>>: attribute lookup <lambda> on __main__ failedlambdaは使えません。モジュールのトップレベルで定義した関数にしてください。
| 渡せる | 渡せない |
|---|---|
| トップレベルの関数 | lambda |
| 数値・文字列・リスト・辞書 | ローカル関数（関数の中で定義） |
| dataclass | 開いているファイル・接続 |
| NumPy配列 | ロック・スレッド |
引数を固定したいならfunctools.partial()を使います。
from functools import partial
ex.map(partial(process, scale=2), items)データの受け渡しコストに注意
大きなデータを渡すと、pickle化・転送・復元に時間がかかります。
with cf.ProcessPoolExecutor() as ex:
results = list(ex.map(process, huge_dataframes)) # 転送コストが大きい並列化で稼いだ時間を転送で食い潰すことがあります。
| 設計 | 効果 |
|---|---|
| 大きなデータを渡す | 遅い |
| ファイルパスを渡して各プロセスで読む | 速い |
| 細かいタスクを大量に投げる | オーバーヘッドが目立つ |
| まとめて渡す（chunksize） | 改善する |
with cf.ProcessPoolExecutor() as ex:
results = list(ex.map(process, items, chunksize=100))chunksizeは何件ずつまとめて子プロセスに渡すかの指定です。1件あたりの処理が軽く件数が多いときに効きます。
グローバル変数は共有されない
import concurrent.futures as cf
counter = 0
def increment(n):
global counter
counter += 1
return counter
if __name__ == "__main__":
with cf.ProcessPoolExecutor(max_workers=2) as ex:
print(list(ex.map(increment, range(4))))
print("親のcounter:", counter)各プロセスは別々のメモリ空間を持ちます。子プロセスで変更しても親には反映されません。
これは欠点ではなく利点でもあります。スレッドのような競合状態やロックの心配がありません。
結果を集めたいなら、戻り値として返してください。
with cf.ProcessPoolExecutor() as ex:
results = list(ex.map(process, items))
total = sum(results)max_workersの決め方
import os
print("CPU数:", os.cpu_count())実行結果：
CPU数: 8既定値はCPU数です。通常はそのままで構いません。
CPU数を超えて増やしても速くなりません。計算する場所が足りないうえ、メモリ消費だけが増えます。プロセスは1つあたりインタプリタ丸ごとなので、スレッドより遥かに重い点に注意してください。
ProcessPoolExecutorのまとめ
ProcessPoolExecutorはCPU処理を本当に並列化できる- 書き方は
ThreadPoolExecutorと同じ - I/O処理では逆に遅い
if __name__ == "__main__":が必須- 関数・引数・戻り値はpickle化できる必要がある（lambdaは不可）
- 大きなデータはパスを渡して各プロセスで読む
- グローバル変数は共有されない——戻り値で返す
max_workersはCPU数が目安
あわせて読みたい
- Python concurrent.futuresとは？並行処理の基本を解説
- Python submit()とmap()の違いは？並行処理の実行方法
- Python as_completed()とは？終わった順に結果を受け取る方法
- Python multiprocessing 関数一覧｜プロセス並列処理で使う関数を解説
- Python functools.partial()とは？引数を固定した関数を作る方法
- Python os.getenv()とは？環境変数を安全に取得する方法
プログラミング学習に興味がある方は、まずは無料体験会へ
