複数csvを高速にdataframeで読み込みたかった話

会社で複数のcsvファイルをdataframeとして読み込みする機会が多いため、処理時間をはかってみる。
テスト用の500行×500列の適当な乱数が入ったcsvを使用。

実行環境

OS:Windows10
CPU:ryzen2700x
メモリ:16GB
python3.6.4

1csvをpd.read_csv読み込み(Cエンジン)

import time
import pandas as pd


def main():
    t1 = time.time()
    # 関数呼び出し
    bench1()

    t2 = time.time()
    elapsed_time = t2-t1
    print(elapsed_time)


# デフォルト
def bench1():
    base_file = r"C:\Users\hoge\Desktop\csv_read_bench\testdata\001.csv"
    df = pd.read_csv(base_file)


if __name__ == '__main__':
    main()

結果(単位は秒)

1回目	2回目	3回目
0.097	0.095	0.097

前座として1ファイルの読み込み速度を計測。
pandasライブラリのread_csv結構速い。

読み込みエンジンをpythonにする

def main():
    t1 = time.time()
    # 関数呼び出し
    bench2()
    t2 = time.time()
    elapsed_time = t2-t1
    print(elapsed_time)


# engineをpythonにする
def bench2():
    base_file = r"C:\Users\hoge\Desktop\csv_read_bench\testdata\001.csv"
    df = pd.read_csv(base_file, engine="python")


if __name__ == '__main__':
    main()

結果(単位は秒)

1回目	2回目	3回目
0.251	0.257	0.258

ファイルパスに日本語が入ってたりするとCエンジンではエラー吐くため
Pythonエンジンで計測。Cエンジンに比べると少し遅くなった。

読み込み列を指定

def main():
    t1 = time.time()
    # 関数呼び出し
    bench3()

    t2 = time.time()
    elapsed_time = t2-t1
    print(elapsed_time)


# 読み込み列を指定
def bench3():
    read_cols = [0, 1, 2]
    base_file = r"C:\Users\hoge\Desktop\csv_read_bench\testdata\001.csv"
    df = pd.read_csv(base_file, engine="python", usecols=read_cols)


if __name__ == '__main__':
    main()

結果(単位は秒)

1回目	2回目	3回目
0.105	0.103	0.104

読み込み列を一部に指定するとそれなりに上がる。
しかし、読み込み列が500列から3列に変わったことで劇的に高速化とはいかないみたいだ。

100csvを読み込ませる

import glob


def main():
    csv_search_path = r"C:\Users\hoge\Desktop\csv_read_bench\testdata\*.csv"
    csv_path_ary = glob.glob(csv_search_path)

    t1 = time.time()
    # 関数呼び出し
    bench4(csv_path_ary)

    t2 = time.time()
    elapsed_time = t2-t1
    print(elapsed_time)


# 100個のcsvをdataframeに格納
def bench4(csv_path_ary):
    for csv_path in csv_path_ary:
        tmp_df = pd.read_csv(csv_path, engine="python")

        if 'df' in locals():
            df = pd.concat([df, tmp_df])
        else:
            df = tmp_df


if __name__ == '__main__':
    main()

結果(単位は秒)

1回目	2回目	3回目
29.95	29.80	29.89

本題、100ファイルを1ファイルずつ読み込んでconcatで結合。
forで処理してるからか、予想通り遅い。

並列化してみる

from multiprocessing import Pool
import os

def main():
    csv_search_path = r"C:\Users\hoge\Desktop\csv_read_bench\testdata\*.csv"
    csv_path_ary = glob.glob(csv_search_path)

    t1 = time.time()
    # 関数呼び出し
    bench5(csv_path_ary)

    t2 = time.time()
    elapsed_time = t2-t1
    print(elapsed_time)


# 並列化してみる
def bench5(csv_path_ary):
    p = Pool(os.cpu_count())
    df = pd.concat(p.map(bench5_multi, csv_path_ary))
    # p.close()


# 並列化読み込み
def bench5_multi(csv_path):
    return pd.read_csv(csv_path, engine="python")


if __name__ == '__main__':
    main()

結果(単位は秒)

1回目	2回目	3回目
4.720	4.658	4.806

multiprocessingで並列化したらだいぶ速くなった。
凄い楽に並列化が出来る。

並列時のCエンジンも計測

from multiprocessing import Pool
import os

def main():
    csv_search_path = r"C:\Users\hoge\Desktop\csv_read_bench\testdata\*.csv"
    csv_path_ary = glob.glob(csv_search_path)

    t1 = time.time()
    # 関数呼び出し
    bench6(csv_path_ary)

    t2 = time.time()
    elapsed_time = t2-t1
    print(elapsed_time)


# 並列化してみる
def bench6(csv_path_ary):
    p = Pool(os.cpu_count())
    df = pd.concat(p.map(bench6_multi_C, csv_path_ary))
    # p.close()


# 並列化読み込み(C engine)
def bench6_multi_C(csv_path):
    return pd.read_csv(csv_path)


if __name__ == '__main__':
    main()

結果(単位は秒)

1回目	2回目	3回目
2.732	2.720	2.722

最後にCエンジンで並列化の時間を計測。
100ファイル読み込むのに3秒なら結構速いのかなぁ。