Beautifulsoup,Mecab,concurrent.futuresを用いたときのメモリー処理

Question

前提・実現したいこと

事前に取得したhtml（４０万程度）からそれぞれbodyを取得し、分かち書きして保存したい。

発生している問題・エラーメッセージ

プログラムにおいて、メモリーが徐々に溜まっていき、インスタンスがダウンしてしまう。
メモリー蓄積を防ぎたい。
しかし、どこが原因かがわからない。

該当のソースコード

import bs4
import MeCab 
import concurrent.futures
import glob
import sys
import os
import gzip
m = MeCab.Tagger('-Owakati')

def _htmlbow(inputs):
  each, i, url = inputs
  print(inputs)

  #既に取っているものはスキップ
  if os.path.exists('%s/%s'%(each,url.replace("./gzhtml/","") )):
    print("already scraped %s"%url)
    return

  #スクレイピングしてgzで保存
  try:
    with gzip.open(url,"rt") as f:
      html = f.read()
    #Bodyの取得
    soup = bs4.BeautifulSoup(html, 'html5lib')
    [s.extract() for s in soup(['style', 'script', '[document]', 'head', 'title'])]
    text = soup.getText()
    #分かち書きして保存
    wakati = m.parse(text).strip()
    with gzip.open('%s/%s'%(each, url.replace("./gzhtml","")), 'wt') as f:
      f.write('{wakati}'.format(wakati=wakati))
    print('finished %d '%(i))
  
  #取得できない場合は失敗リストに保存
  except Exception as e:
    with open("failed_scraping.txt","a") as f:
      f.write('{url}\n'.format(url=url.replace("./gzhtml/","")))
    print('failed     %d '%(i))


def htmlbow():
  #htmlリストを読み込み、インデックスを振る
  urls = []
  for i,ents in enumerate(glob.glob("./gzhtml/*")):
    urls.append(["bow",i,ents])
  print('load finished')
  #分散処理
  with concurrent.futures.ProcessPoolExecutor(max_workers = 992) as executor:
    executor.map(_htmlbow, urls)


"""htmlbow"""
if '--htmlbow' in sys.argv:
   htmlbow()

Answer 1

with concurrent.futures.ProcessPoolExecutor(max_workers = 992) as executor:

max_workersを設定しないようにしても問題は発生しますか？

あと例外メッセージが発生してるなら、質問文に追記していただくと回答時に参考情報になります。

---

追記
以下のように変更してくださいな。

with concurrent.futures.ProcessPoolExecutor(max_workers = 992) as executor:

↓

with concurrent.futures.ProcessPoolExecutor() as executor:

---

maxworker 指定なしで大丈夫でした。

指定がない場合はプロセス数はどのように処理されているのでしょうか？

Githubにある、CPythonのProcessPoolExecutorの__init__部分より引用。

if max_workers is None:
    self._max_workers = os.cpu_count() or 1

■参考情報
os.cpu_count()

■余談
1,プロファイルを取ってみたほうがいいと思いますが。
URLからダウンロードしている部分 with gzip.open(url,"rt") はThreadPoolExecutorでI/O をオーバーラップしたほうが良いかもしれません。

2,max_workers = 992に関して
この設定値はどこかのサイトのサンプルソースの設定値なのでしょうか？

最大ワーカープロセス数=992なため
a,プロセス数の生成が最大で992個行われる
b,1プロセスあたり20MBメモリを消費

このプログラムは最大で992*20MBの19,840MB(大凡19.8GB)のメモリを使用します。

---

一番重要な、質問文のメモリ問題は解決したのでしょうか？
その点気になってますが。。。

以下はコメント欄への質問への回答。
できるだけ調べたことを書いてくださいな。どのような内容を調べて、どこの記述がわからないのか。丸投げではなく、そういう形の質問にしてくださいな。

ProcessPoolExecutorとThreadPoolExecutor違い

|Executor名|並列対象|オーバーラップ対象|GILの制限|データの受け渡し|
|:--|:--:|--:|
|ProcessPoolExecutor|プロセス|マルチプロセスなため、CPUの計算|影響を受けない|pickle可能なオブジェクト|
|ThreadPoolExecutor|スレッド|IOがボトルネックな処理(ネットワークやDisk)|影響を受ける|制約なし|

GILはGlobalInterpreterLockといいます。

Comments

[dialectic4th]

以下のように試しみましたが、状況変わらずです。
```
with concurrent.futures.ProcessPoolExecutor(max_workers = 992) as executor:
for url in urls:
executor.submit(_htmlbow, url)
```

[dialectic4th]

コメント先間違えてましたね・・・

maxworker 指定なしで大丈夫でした。
指定がない場合はプロセス数はどのように処理されているのでしょうか？

[dialectic4th]

ProcessPoolExecutorとThreadPoolExecutorの違いが、わかっていないのですが、どういう違いがあるのでしょうか？

Answer 2

executor.mapの部分じゃないかなぁと推測します。代わりにfor文で一件ずつexecutor.submitすると現象が改善したりしませんか？