concurrent.futuresで関数内部の関数をsubmitしたい

Question

# data_predicting.py

from tqdm import tqdm
import concurrent.futures 

class PredictData:
  def __init__(self,X_train_trans,X_test_trans,y_train,y_test):
    self.X_train_trans=X_train_trans
    self.X_test_trans=X_test_trans
    self.y_train=y_train
    self.y_test=y_test
  def GraForestCl_predict_data(self):
    def forcount(random_st):
      print('start')
      test_score=0
      train_score=0
      n_es=0
      max=0
      rate=0
      for i in tqdm(range(1,3)):
        for j in tqdm(range(1,3)):
          for k in tqdm(np.arange(.01,.03,.01)):
            forest=GradientBoostingClassifier(random_state=random_st,n_estimators=i,max_depth=j,learning_rate=k)
            forest=forest.fit(self.X_train_trans,self.y_train)
            if forest.score(self.X_test_trans,self.y_test) > test_score:
              test_score=forest.score(self.X_test_trans,self.y_test)
              train_score=forest.score(self.X_train_trans,self.y_train)
              n_es=i
              max=j
              rate=k
      return test_score,train_score,n_es,max,rate
    with concurrent.futures.ProcessPoolExecutor(None) as executor:
      print(forcount(0)) # 正常に作動し、値(test_score,,,rate)が表示される
      print('開始')
      futures=[executor.submit(forcount,i) for i in range(2)] # ここが問題
      print('途中')
      for future in concurrent.futures.as_completed(futures):
        # test_score,train_score,n_es,max,rate=future.result()
        # print(f'GraForestCl_Train Score n_es{n_es},max{max},rate{rate}:{train_score}')
        # print(f'GraForestCl_Test Score n_es{n_es},max{max},rate{rate}:{test_score}')
        print(future.result())
        print()

if __name__=='__main__':
  prd=PredictData(X_train_trans,X_test_trans,y_train,y_test)
  prd.GraForestCl_predict_data()
'''
以下実行結果

開始
途中
print(future.result())の部分で
AttributeError:Can't pickle local object 
'PredictData.GraForestCl_predict_data.<locals>.forcount'

'''

GraForestCl_predict_data関数の中のforcount関数を並行処理したいのですが、予想ですがsubmitされておらずforcount関数が作動しません。
助言をお願いします。

Answer 1

executor.submit(forcount,i) は実行する関数を追加したあと、ただちに終了するので、すべての実行の完了を待機するのは、concurrent.futures.as_completed(futures) の部分になります。
そのため、質問に記載されている「本来なら「開始」と「途中」の間に「start」が２個表示されるはず」とはなりませんが、並列実行されています。

Python - concurrent.futures を使った並列化の方法について

追記

ProcessPoolExecutor を使う場合、並列実行させる関数を関数内に定義するのはダメみたいです。外側に定義しましょう。

NG

def GraForestCl_predict_data(self):
    def func():
        ...

OK

def GraForestCl_predict_data(self):
        ...

def func(self):
    ...

---

import concurrent.futures

from sklearn.datasets import make_hastie_10_2
from sklearn.ensemble import GradientBoostingClassifier


class PredictData:
    def __init__(self, X_train_trans, X_test_trans, y_train, y_test):
        self.X_train_trans = X_train_trans
        self.X_test_trans = X_test_trans
        self.y_train = y_train
        self.y_test = y_test

    def forcount(self):
        # この中身は適宜編集してください。以下はテスト用のコードです。
        clf = GradientBoostingClassifier(n_estimators=100, learning_rate=1.0, max_depth=1).fit(
            X_train, y_train
        )
        score = clf.score(X_test, y_test)

        return score

    def GraForestCl_predict_data(self):
        num_tasks = 2  # タスクの合計数
        num_workers = 8  # 並列実行数

        with concurrent.futures.ProcessPoolExecutor(num_workers) as executor:
            # タスクをプロセスプールに追加する。
            futures = [executor.submit(self.forcount) for i in range(num_tasks)]

        # 実行の完了を待機し、結果を取得する。
        results = [x.result() for x in futures]
        print(results)


# データセットを作成する。
X, y = make_hastie_10_2(random_state=0)
X_train, X_test = X[:2000], X[2000:]
y_train, y_test = y[:2000], y[2000:]

prd = PredictData(X_train, X_test, y_train, y_test)
prd.GraForestCl_predict_data()

Comments

[hima-mura.]

ではforcount関数自体はexecutorに渡せており、
正常に動いた場合「途中」のつぎに「start」がくるということでしょうか？
理解及ばず申し訳ないのですがどこの部分が間違っているのかわかりません。

[tiitoi]

submit() は関数を開始したら、submit() そのものはすぐに終わるので、開始->途中 -> start -> start という順番で出力が出ていても正常です。この3つは別々に実行されるため、2つの「start」と「途中」がどの順番で出力されるかはランダムです。関数が並列で動いているか確認したい場合は関数の中に適当な print 文を入れて確認すればいいのではないでしょうか。

[hima-mura.]

そもそも並列以前にAttributeErrorで止まってしまってforcount関数が正常に実行できていないので、何回実行しても「開始」「途中」しか表示されないのが問題なのですが。

[tiitoi]

だとすると、forcount 内のコードが問題ないのではないでしょうか。StackTrace を含めて、エラー全部を貼っていただけますか。

[hima-mura.]

ありがとうございます。
上記のコードの一番下のコメントアウト''' '''のところに載せておきました。

[tiitoi]

すいません。StackTrace を知りたいのは
Traceback (most recent call last) の下にそのエラーが発生するまでの履歴が出ていると思うので、それを貼ってくださいという意味です。

[hima-mura.]

AttributeError
Can't pickle local object 'PredictData.GraForestCl_predict_data.<locals>.forcount'
File data_predicting.py, line 65, in GraForestCl_predict_data
print(future.result())
File data_predicting.py, line 76, in <module>
prd.GraForestCl_predict_data()

[tiitoi]

エラーの原因がわかりました。並列実行する関数 forcount() を関数内で定義しているのが問題です。関数の外側で定義するようにしましょう。

回答に動作するサンプルコードを追記しました。

[hima-mura.]

できればタイトル通り、GraForestCl_predict_data関数内で定義した関数であるforcountをうまく処理したかったのですが、やっぱりむりみたいですね。

GraForestCl_predict_data関数のスコープ外にコードしたら正常に動きました。ありがとうございました。

[hima-mura.]

別の問題がはっせいしてしまいました。また改めてスレを貼るので時間があるときに助言をくださると助かります。