[Python] 並列処理でCPUのコア数を制限する方法

Pythonで並列処理を行う際、CPUのコア数を制限するには、concurrent.futuresモジュールのProcessPoolExecutorやmultiprocessingモジュールを使用します。

ProcessPoolExecutorでは、max_workers引数に制限したいコア数を指定します。

multiprocessingモジュールでは、Poolクラスのprocesses引数にコア数を指定することで制限できます。

これにより、指定した数のプロセスのみが並列に実行され、CPUの過負荷を防ぐことができます。

concurrent.futuresを使った並列処理

concurrent.futuresモジュールの概要

concurrent.futuresモジュールは、Pythonで並列処理を簡単に実装するための高レベルなインターフェースを提供します。

このモジュールは、スレッドやプロセスを使った非同期処理をサポートしており、特にCPUバウンドなタスクやI/Oバウンドなタスクに対して効果的です。

主にThreadPoolExecutorとProcessPoolExecutorの2つのクラスが用意されています。

ProcessPoolExecutorの基本的な使い方

ProcessPoolExecutorは、プロセスを使用して並列処理を行うためのクラスです。

これにより、CPUコアを最大限に活用することができます。

基本的な使い方は以下の通りです。

from concurrent.futures import ProcessPoolExecutor
def task(n):
    return n * n
# Windowsで実行する場合は、if __name__ == '__main__':の中に
# 記述しないとエラーが発生する
if __name__ == '__main__':
    with ProcessPoolExecutor() as executor:
        results = list(executor.map(task, range(10)))
    print(results)

このコードでは、0から9までの数を二乗するタスクを並列に実行しています。

max_workersでコア数を制限する方法

ProcessPoolExecutorでは、max_workers引数を使用して同時に実行するプロセスの数を制限できます。

これにより、CPUコアの数に応じた最適な並列処理が可能になります。

以下のように指定します。

with ProcessPoolExecutor(max_workers=4) as executor:
    results = list(executor.map(task, range(10)))

この例では、最大4つのプロセスが同時に実行されます。

実際のコード例：コア数を制限した並列処理

以下は、max_workersを使用してコア数を制限した実際のコード例です。

from concurrent.futures import ProcessPoolExecutor
import time
def compute_square(n):
    time.sleep(1)  # 模擬的な処理時間
    return n * n
if __name__ == "__main__":
    start_time = time.time()
    with ProcessPoolExecutor(max_workers=2) as executor:
        results = list(executor.map(compute_square, range(5)))
    print("結果:", results)
    print("処理時間:", time.time() - start_time)

このコードでは、5つの数の二乗を計算するタスクを2つのプロセスで並列に実行しています。

出力結果は以下のようになります。

結果: [0, 1, 4, 9, 16]
処理時間: 約3秒

ThreadPoolExecutorとの違い

ThreadPoolExecutorはスレッドを使用して並列処理を行いますが、ProcessPoolExecutorはプロセスを使用します。

以下の点が主な違いです。

このように、タスクの性質に応じて適切なExecutorを選択することが重要です。

multiprocessingモジュールを使った並列処理

multiprocessingモジュールの概要

multiprocessingモジュールは、Pythonで並列処理を実現するための標準ライブラリです。

このモジュールは、複数のプロセスを生成し、それぞれが独立して実行されるため、CPUバウンドなタスクに特に効果的です。

multiprocessingを使用することで、Pythonのグローバルインタプリタロック(GIL)の制約を回避し、マルチコアCPUを最大限に活用できます。

Poolクラスの使い方

Poolクラスは、複数のプロセスを管理し、タスクを並列に実行するための便利なインターフェースを提供します。

基本的な使い方は以下の通りです。

from multiprocessing import Pool
def square(n):
    return n * n
if __name__ == "__main__":
    with Pool() as pool:
        results = pool.map(square, range(10))
    print(results)

このコードでは、0から9までの数を二乗するタスクを並列に実行しています。

processes引数でコア数を制限する方法

Poolクラスでは、processes引数を使用して同時に実行するプロセスの数を制限できます。

これにより、CPUコアの数に応じた最適な並列処理が可能になります。

以下のように指定します。

with Pool(processes=4) as pool:
    results = pool.map(square, range(10))

この例では、最大4つのプロセスが同時に実行されます。

実際のコード例：multiprocessingでのコア数制限

以下は、processesを使用してコア数を制限した実際のコード例です。

from multiprocessing import Pool
import time
def compute_square(n):
    time.sleep(1)  # 模擬的な処理時間
    return n * n
if __name__ == "__main__":
    start_time = time.time()
    with Pool(processes=2) as pool:
        results = pool.map(compute_square, range(5))
    print("結果:", results)
    print("処理時間:", time.time() - start_time)

このコードでは、5つの数の二乗を計算するタスクを2つのプロセスで並列に実行しています。

出力結果は以下のようになります。

結果: [0, 1, 4, 9, 16]
処理時間: 約3秒

multiprocessingとconcurrent.futuresの違い

multiprocessingとconcurrent.futuresはどちらも並列処理を実現するためのモジュールですが、いくつかの違いがあります。

このように、タスクの性質や使用する環境に応じて、適切なモジュールを選択することが重要です。

並列処理のパフォーマンス最適化

コア数を制限する理由とメリット

コア数を制限することにはいくつかの理由とメリットがあります。

主な理由は以下の通りです。

• リソースの最適化: CPUコアを過剰に使用すると、コンテキストスイッチが増え、オーバーヘッドが発生します。

適切なコア数を設定することで、リソースを効率的に使用できます。

• メモリ使用量の管理: 各プロセスは独自のメモリ空間を持つため、過剰なプロセスを生成するとメモリ使用量が増加します。

コア数を制限することで、メモリの消費を抑えられます。

• 安定性の向上: 過剰な並列処理は、システムの安定性を損なう可能性があります。

適切なコア数を設定することで、システム全体のパフォーマンスを向上させることができます。

過剰な並列処理によるデメリット

過剰な並列処理には以下のようなデメリットがあります。

• コンテキストスイッチの増加: プロセス間での切り替えが頻繁になると、CPUのオーバーヘッドが増加し、全体のパフォーマンスが低下します。
• リソースの競合: 同時に多くのプロセスがリソースを要求すると、競合が発生し、待機時間が増加します。
• デバッグの難しさ: 並列処理が多すぎると、エラーの発生源を特定するのが難しくなります。

これにより、開発や保守が複雑になります。

適切なコア数の選び方

適切なコア数を選ぶためのポイントは以下の通りです。

• タスクの性質: CPUバウンドなタスクの場合、コア数をCPUの物理コア数に合わせるのが理想です。

一方、I/Oバウンドなタスクでは、コア数を増やすことでパフォーマンスが向上することがあります。

• システムのリソース: 使用するシステムのメモリやCPUの性能を考慮し、リソースに見合ったコア数を設定します。
• 実験と調整: 異なるコア数でのパフォーマンスを測定し、最適な設定を見つけるために実験を行います。

ベンチマークテストを実施することが有効です。

CPU負荷のモニタリング方法

CPU負荷をモニタリングする方法はいくつかあります。

以下は一般的な方法です。

• topコマンド: LinuxやmacOSで使用できるコマンドラインツールで、リアルタイムでCPU使用率を確認できます。
• psutilライブラリ: PythonでCPU使用率をプログラム的に取得するためのライブラリです。

以下のように使用します。

import psutil
cpu_usage = psutil.cpu_percent(interval=1)
print(f"CPU使用率: {cpu_usage}%")

• システムモニタリングツール: WindowsのタスクマネージャーやLinuxのhtopなど、GUIベースのツールを使用してCPU負荷を視覚的に確認できます。

これらの方法を活用して、システムのCPU負荷を適切にモニタリングし、並列処理のパフォーマンスを最適化することが重要です。

応用例：並列処理の実践的な活用

大量データの処理における並列処理の活用

大量のデータを処理する際、並列処理を活用することで処理時間を大幅に短縮できます。

例えば、データベースからのデータ取得や、データの前処理、集計などのタスクを並列に実行することが可能です。

以下は、multiprocessingを使用して大量データを並列処理する例です。

from multiprocessing import Pool
import pandas as pd
def process_data(chunk):
    # データ処理のロジック
    return chunk.sum()
if __name__ == "__main__":
    data = pd.DataFrame({'value': range(1000000)})
    chunks = np.array_split(data, 10)  # データを10分割
    with Pool(processes=4) as pool:
        results = pool.map(process_data, chunks)
    total = sum(results)
    print("合計:", total)

このコードでは、データを10分割し、4つのプロセスで並列に処理しています。

Webスクレイピングでの並列処理

Webスクレイピングでは、複数のページを同時に取得することで、全体の処理時間を短縮できます。

concurrent.futuresのThreadPoolExecutorを使用することで、簡単に並列処理を実現できます。

以下はその例です。

import requests
from concurrent.futures import ThreadPoolExecutor
def fetch_url(url):
    response = requests.get(url)
    return response.text
urls = ['http://example.com/page1', 'http://example.com/page2', 'http://example.com/page3']
with ThreadPoolExecutor(max_workers=3) as executor:
    results = list(executor.map(fetch_url, urls))
for result in results:
    print(result[:100])  # 各ページの最初の100文字を表示

このコードでは、3つのURLを並列に取得しています。

機械学習モデルのトレーニングでの並列処理

機械学習モデルのトレーニングは計算負荷が高いため、並列処理を活用することでトレーニング時間を短縮できます。

特に、ハイパーパラメータのチューニングや、複数のモデルを同時にトレーニングする際に有効です。

以下は、joblibライブラリを使用した例です。

from joblib import Parallel, delayed
from sklearn.ensemble import RandomForestClassifier
from sklearn.datasets import make_classification
X, y = make_classification(n_samples=10000, n_features=20)
def train_model(n_estimators):
    model = RandomForestClassifier(n_estimators=n_estimators)
    model.fit(X, y)
    return model.score(X, y)
results = Parallel(n_jobs=4)(delayed(train_model)(n) for n in [10, 50, 100, 200])
print("モデルの精度:", results)

このコードでは、異なる数の決定木を持つランダムフォレストモデルを4つのプロセスで並列にトレーニングしています。

並列処理を使ったファイルI/Oの最適化

ファイルの読み書きはI/Oバウンドな処理であり、並列処理を活用することで効率的に行うことができます。

特に、大量のファイルを同時に処理する場合に効果的です。

以下は、multiprocessingを使用してファイルを並列に読み込む例です。

from multiprocessing import Pool
import os
def read_file(file_path):
    with open(file_path, 'r') as f:
        return f.read()
if __name__ == "__main__":
    file_paths = ['file1.txt', 'file2.txt', 'file3.txt']
    with Pool(processes=3) as pool:
        results = pool.map(read_file, file_paths)
    for content in results:
        print(content[:100])  # 各ファイルの最初の100文字を表示

このコードでは、3つのファイルを並列に読み込んでいます。

これにより、I/O待ち時間を短縮し、全体の処理時間を短縮できます。

よくある質問

まとめ

この記事では、Pythonにおける並列処理の基本的な概念から、具体的な実装方法、パフォーマンス最適化のポイント、実践的な応用例まで幅広く取り上げました。

特に、concurrent.futuresやmultiprocessingモジュールを活用することで、CPUやメモリのリソースを効率的に利用し、処理時間を短縮する方法について詳しく解説しました。

これを機に、並列処理を活用して自分のプロジェクトやタスクのパフォーマンスを向上させることに挑戦してみてはいかがでしょうか。