ProcessPoolExecutor を利用した並列処理について

Question

ProcessPoolExecutor を利用して並列処理を実行する際、
並列処理を行う自作関数で引数が1つ以上ある場合、どのように指定すればよいでしょか。

たとえば、

def 自作関数(引数1,引数2):
    処理内容

def main():
    with concurrent.futures.ProcessPoolExecutor() as executor:
        for result, test in zip(executor.map(自作関数, test),test):
            

の場合、自作関数の引数2などにイテラブルなtest以外のデータを渡したいです。

ご教授いただけると助かります。
よろしくお願い致します。

Answer 1

XxxPoolExecutorに限らないmapの使い方の話ですね。

def 自作関数(引数1, 引数2):
    処理内容

とある場合には、引数リスト1 = [1, 2, 3]と引数リスト2 = [4, 5, 6]を準備して、

for r in map(自作関数, 引数リスト1, 引数リスト2):
    ....

と書けば、自作関数(1, 4)->自作関数(2, 5)->自作関数(3, 6)と順次よばれて、その各戻り値rを使用してFOR文を回せます。

追記

あとよくある、引数1はリストだけど引数2は固定値、という場合は以下のように書きます。※functools.partialではCで書かれた関数の第2引数固定ができなかったりするので、itertools.repeatのほうが汎用的にどこでも使えます。

引数2の固定パターン1: functools.partial

from functools import partial

ループ用関数 = partial(自作関数, 引数2=固定値)
for r in map(ループ用関数, 引数リスト1):
    ....

引数2の固定パターン2: itertools.repeat

from itertools import repeat

for r in map(自作関数, 引数リスト1, repeat(固定値)):
    ....

Comments

[NoPython_Nolife]

非常に分かりやすいご説明ありがとうございます！大変助かりました！

Answer 2

渡したい引数がpickle可能なオブジェクトならmapに引数を追加したら渡せますが、やりたいことはこれ以外ですか？

# -*- coding: utf8 -*-
from os import getpid
from time import sleep
from threading import current_thread
import concurrent.futures

import numpy as np


def 自作関数(引数1, 引数2):
    sleep(3)
    return f"pid={getpid()}, tid={current_thread().ident}, {引数1}, {引数2}"


def main():
    np.random.seed(42)
    test = np.random.random_integers(0, 500, 30)
    test2 = np.random.random_integers(0, 500, 30)
    assert len(test) == len(test2)
    with concurrent.futures.ProcessPoolExecutor() as executor:
        for result, test in zip(executor.map(自作関数, test, test2), test):
            print(result)


if __name__ == '__main__':
    main()

Comments

[NoPython_Nolife]

非常に的確なご案内ありがとうございます！仰るお通り、まさにこの内容を求めておりました。大変助かりました！ありがとうございました！

Answer 3

引数をタプルにまとめください。

def your_func(*args):

Comments

[NoPython_Nolife]

勉強になります！ありがとうございます！