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

Question

### 
環境
windows10、 Vscode、  python 2.88

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

###

```python
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秒くらいでした。

Answer

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

Answer

以下、実行環境です。

```bash
$ 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` の中身はリストのリストになっています

```python
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')
```

Answer

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

```Python
import pandas as pd

df = pd.read_csv(file_path)

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

Answer

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

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

関連した質問