Pythonでの並列処理：multiprocessing.Pool() の使い方

概要

Pythonで大規模データ処理や重い計算処理を行う際に、シングルスレッドでは処理速度がボトルネックになることがあります。そこで役立つのが、multiprocessing モジュールです。この記事では、その中でも特に使い勝手が良い multiprocessing.Pool() を使った並列処理の方法についてメモします。

multiprocessing.Pool() とは？

multiprocessing.Pool() は、複数のプロセスを使って並列にタスクを実行するための機能です。これを使うことで、同時に複数の処理を効率よく実行できるようになります。

基本的な使い方

multiprocessing.Pool() を使うときは、通常以下のような構造で実装します。

import multiprocessing

def タスク関数(引数):
    # 処理の内容
    return 処理結果

if __name__ == "__main__":
    with multiprocessing.Pool(プロセス数) as pool:
        結果 = pool.map(タスク関数, 引数のリスト)

• multiprocessing.Pool(プロセス数) は、同時に実行するプロセスの数を指定します。
• pool.map() は、リストなどのデータに対して並列処理を適用するための関数です。

multiprocessing.Pool() を使った実例

実際に具体的なコードでmultiprocessing.Pool()の使ってみる。ここでは、ファイルを並列処理で読み込むシミュレーションをします。

import multiprocessing
import time

# ファイル処理などの重い処理を模倣する関数
def process_file(file_name):
    print(f"{file_name} の処理開始")
    time.sleep(2)  # ファイル処理に時間がかかることをシミュレート
    print(f"{file_name} の処理完了")
    return f"{file_name} の結果"

if __name__ == "__main__":
    # 処理したいファイル名のリスト
    file_list = ["file1.txt", "file2.txt", "file3.txt", "file4.txt", "file5.txt"]

    # Poolを使って並列処理
    with multiprocessing.Pool(processes=3) as pool:  # プロセス数を3に設定
        results = pool.map(process_file, file_list)  # ファイルリストを並列処理
    
    # 処理結果を表示
    print("すべてのファイル処理が完了しました:")
    for result in results:
        print(result)

• multiprocessing.Pool(processes=3): 3つのプロセスを使ってファイルを並列処理しています。システムのコア数に合わせて適切なプロセス数に設定する必要があります。
• with 文: with を使うことで、リソース管理が自動化され、明示的に close() や terminate() する必要がなく、ブロックが終了した時点で自動的にリソースを解放します。
• pool.map(): リストの各要素に対して並列処理を適用します。この例では、ファイル名のリストに対して処理関数 process_file() を実行しています。

実行結果

実行すると、以下のようにファイル処理が並列に行われる様子が確認できます。

file1.txt の処理開始
file2.txt の処理開始
file3.txt の処理開始
file1.txt の処理完了
file4.txt の処理開始
file2.txt の処理完了
file5.txt の処理開始
file3.txt の処理完了
file4.txt の処理完了
file5.txt の処理完了
すべてのファイル処理が完了しました:
file1.txt の結果
file2.txt の結果
file3.txt の結果
file4.txt の結果
file5.txt の結果

大規模データの処理

multiprocessing.Pool() は、大量のデータや複数のファイルを並列に処理する場合に非常に役立ちます。これはよく現場で使用されていると思います。

def process_data(data_chunk):
    # 複雑なデータ処理
    return 処理結果

if __name__ == "__main__":
    data_list = 大量データリスト

    with multiprocessing.Pool() as pool:
        results = pool.map(process_data, data_list)
    
    # ここで結果をまとめたり、ファイルに出力

このようにデータ分析系のシステムではよく使用されます。