python の並列化により、行変換処理の速度を上げたい

Question

環境
windows10、 Vscode、 python 2.88

pythonで、下記のようにCSVファイルから行単位のデータを取り込んで
各行を並列に処理し、速度を上げたいと考えました。
threadとprocessでmapにより並列化しましたが、普通にloop処理した方が
10倍近く早く、逆に大幅に遅くなりました。
速度を上げたいのですが、どのような方法があるでしょうか？

import time
from concurrent.futures import ThreadPoolExecutor
# from concurrent.futures import ProcessPoolExecutor

def list_app(row):
    d_row = row.split(',')
    row = row + ',' + d_row[5] + ',' + d_row[6] + ',' + d_row[7] 
    return row 

if __name__=='__main__':
    
    d_f_con=[]
    thread_flag = True
    
    with open('C:/Users/fxscX0cd6/Documents/test.csv','r',encoding='shift-jis') as f:
        f_con=f.readlines()
        
    startt = time.time()
    
    if thread_flag == False:
        for i,row in enumerate(f_con):
            d_row = row.split(',')
            row = row + ',' + d_row[5] + ',' + d_row[6] + ',' + d_row[7] 
            d_f_con.append(row) 
        
    else:
        with ThreadPoolExecutor(max_workers=10) as executor:
        # with ProcessPoolExecutor(max_workers=10) as executor:
            results = executor.map(list_app, f_con)
            results = list(results) 
            
    endt=time.time() -startt
    
    print(endt)

threadとprocess仕様で１秒くらい、通常loopで0.1秒くらいでした。

Comments

[bsdfan]

速度を上げたい理由はなんでしょう？
・ファイルが大量にある
・実際のファイルのサイズはもっと大きい
・実際はもっと複雑な処理をしていて時間がかかっている
それによって、とれる方法が違ってくると思います。

[jbpb0]

> threadとprocess仕様で１秒くらい

マルチスレッドよりもマルチプロセスの方が、もっと遅くなりませんか？
(プロセスの生成の方が、スレッドの生成よりも時間がかかるため)

google colabで、

xxx = np.random.rand(100000, 10)
np.savetxt('test.csv', xxx, delimiter=',')

で作ったcsvファイルを使って質問のコードを実行したら、実行時間は下記の通りでした

並列化無し：0.2秒くらい
マルチスレッド：3秒くらい
マルチプロセス：30秒くらい

[quickquip]

Pythonのバージョンはなにかの書き間違いですか? （それとも本当にPython2系を使っている? VS Codeのプラグインのバージョンを書いてしまったとかもありえる?）
追加: concurrent がある以上2系なはずはなかったですね

Answer 1

pandas を使用してみてはいかがでしょうか。
for文より高速に処理できます。
以下のようなコードで、list_app関数に近い処理ができると思います。

import pandas as pd

df = pd.read_csv(file_path)

df_list_app = df[['col_5', 'col_6', 'col_7']]

Comments

[kazuya_]

ご回答ありがとうございます。
実際には、splitでなくて正規表現を作成する処理なのですが、
たぶん１行に数秒かかっている可能性があります。
pandasは確か、C++で作成されているんですかね。
split等使用時には、参考にさせていただきます。

[fpfpfp]

extract メソッドが使えるかもしれません
https://note.nkmk.me/python-pandas-str-extract-extractall/

また、正規表現の書き方も計算速度に大きく関係します
https://blog.shin1x1.com/entry/regex-performance

ここまで書いて、『正規表現を作成する処理』なので関係なさそうですね。。

[kazuya_]

ご連絡ありがとうございました。
参考にさせていただきます。
文字列を正規表現に置き換える処理を行っています。

Answer 2

並列処理をする場合には、プロセスやスレッドの準備や生成・削除など、並列処理をするために追加で必要な処理の時間(オーバーヘッド)がかかります。処理の内容や分割方法などによってはそのオーバーヘッドほ方が大きくなってしまう場合があります。

今回の例では、処理内容そのものが時間がかからないものなので、並列化の恩恵が受けられないということでしょう。

Comments

[kazuya_]

ご回答ありがとうございます。
わかりにくい質問で申し訳ありません。
オーバーヘッドについては、それが原因だと気づいております。
ただ、メモリ上に展開していますので、ここまで差があると使い物にならないような気がします。
コーディングの工夫(thread系の別のライブラリの使用)で、もう少し何とかならないものでしょうか？

[TakaiY]

ならないでしょう。

使いものにならないということはまったくありません。
たとえば、1行の処理に10分かかるような処理を1000行分やる場合、直列であれば10000分(2.8時間)かかるところを並列でやれば10分の1とか100分の1の時間で済みます、オーバーヘッドは30秒かかりますが影響は無視できます、というのが並列処理です。 充分な恩恵です。

[kazuya_]

そうですね。オーバーヘッドの時間がずっと一定だとして、１行で数秒の処理ですと
恩恵があるかもしれません。少し試してからまた、投稿いたします。

[kazuya_]

１行３秒の処理で44行をLOOPしてみました。
結果、threadを使用しない場合128秒に対して、max_workersを10から5とかに変化させましたが、
約125から127秒となり、ほぼ差がありませんでした。
このことから、concurrent.futuresの使用では恩恵がほぼ無いようです。
processのほうは、正規表現の関数を使用したためか、エラーになったのでやっていません。

[TakaiY]

情報ありがとうございます。

その 1行3秒の処理というのはどういった処理でしょう。pythonにはGIL(global interpreter lock)という制約がありまして、処理内容によってはマルチスレッドで高速化しにくいのです。

[kazuya_]

なるほど
re.searchで検索した結果を、ディクショナリ配列に格納していく処理を処理時間延長の為
10000回LOOPしています。
どうやら、排他ロックを繰り返しながら、1行づつ実行している状態のようです。
ファイルの読み込み等のI/O関係か、あるいはGILのないライブラリを使用する方法に限定されてしまいますね

[TakaiY]

そうですね。 もしくは、マルチプロセスにするか。

Answer 3

GILを無効化したpythonで実行したら1/2程度まで速くなりました。
https://github.com/colesbury/nogil

Answer 4

以下、実行環境です。

$ lsb_release -d
Description:    Ubuntu 22.04.1 LTS

$ uname -isr
Linux 5.15.0-22-generic x86_64

$ cat /proc/cpuinfo | grep model.name | head -1
model name	: Intel(R) Core(TM) i5-6600K CPU @ 3.50GHz

# CPU 数
$ nproc
6

# CSV ファイルの行数(1億行)
$ wc -l test.csv
100000000 test.csv

質問に掲載されているコードで thread_flag = False とした場合の実行時間は、55.18664 sec になりました。

以下のコードでは ProcessPoolExecutor() を使っていて、CSV ファイルを 192 分割、6 個のプロセスで並列処理しています。実行時間は 18.27044 sec でした。
※ result の中身はリストのリストになっています

import os
import concurrent.futures as cf
from time import time

def list_app(cbegin, cend):
    global f_con
    splitted = []
    for r in f_con[cbegin:cend]:
        row = r.rstrip()
        d_row = row.split(',')
        row += ',' + d_row[5] + ',' + d_row[6] + ',' + d_row[7]        
        splitted.append(row)

    return splitted

def app(lines, workers, nchunks):
    chunksize = lines // nchunks
    with cf.ProcessPoolExecutor(max_workers=workers) as executor:
        cbegin = [chunksize * i for i in range(nchunks)]
        cend = cbegin[1:] + [lines]
        result = executor.map(list_app, cbegin, cend)

    return result

if __name__ == '__main__':
    with open('test.csv', 'r') as f:
        f_con = f.readlines()

    lines = len(f_con)
    workers = os.cpu_count()
    nchunks = workers * 32

    start = time()
    result = app(lines, workers, nchunks)
    result = list(result)
    duration = time() - start

    print(f'{duration:.5f} sec')