LoginSignup
2
2
@side_job_acc(たいやき)

Pythonで大容量CSVファイルを分割してみた。

Last updated at Posted at 2024-03-12

作成背景

仕事中に数百万行あるCSVがMicrosoftExcelやメモ帳で開けなかったのでPythonを使って分割してみました。
そもそも一度に出力せず作成の段階で分割しといてほしかったのですがそうも言っていらないのが世の中というものです。

全体

今回作成したソースコードの全体となります。
loggerを使っていなかったり実験的にpandasではなくdaskを使ったのは個人的なスクリプトだったのでめんどくささと遊びが勝ちました。

作成コード 
import argparse
import dask.dataframe as dd
from pathlib import Path
import sys

#引数の受け取り
parser = argparse.ArgumentParser()  # クラスの作成
parser.add_argument("Path") # 第一引数(必須)
parser.add_argument("-c", "--Chunk", type=int, default=1000)  # オプション引数->デフォルト1000
parser.add_argument("-e", "--Encoding", default="cp932")  # オプション引数->デフォルトcp932

args = parser.parse_args()


#ディレクトリ(ファイル)チェック
def checkPath(target_path):
    # 1.ディレクトリであれば配下のCSVファイルを検索しリストを返却
    # 2.CSVであればリストに格納して変換
    # 3.それ以外なら終了
    if target_path.is_dir():
        print("ディレクトリが見つかりました。")
        return target_path.glob('*.csv')
    elif target_path.is_file() and target_path.suffix == ".csv":
        print("CSVファイルが見つかりました。")
        return [target_path]
    else:
        print(f"{target_path}は存在しない、もしくはCSVファイルではない可能性があります。")
        sys.exit()

# ディレクトリ作成
def createDir(target_path):
    # ディレクトリが存在しなければ作成
    # ※なんとなく作っただけなのでメイン処理に埋め込んでもいい
    flg = False
    if not target_path.exists():
        target_path.mkdir()
        flg = True
    return flg

# 主処理
def main():

    # 引数の受け取り
    path = Path(args.Path)
    chunk_size = args.Chunk

    # CSVファイルリストの作成
    file_list = checkPath(path)

    # ファイルリストをループして処理
    for file_path in file_list:
        parent_dir = Path.joinpath(file_path.parent, file_path.stem)
        create_flg = createDir(parent_dir)

        # ディレクトリの作成ができたら出力->別に要らない処理
        if create_flg:
            print(f"ディレクトリを作成しました。->{parent_dir}")
        
        print(f"ファイル読み込み開始->{file_path}")

        # ファイル読み込み
        ddf = dd.read_csv(file_path, encoding=args.Encoding, header=0, dtype="object").compute()

        # 分割数の計算
        chunk_count = len(ddf) // chunk_size + 1

        # ループ処理で分割
        for i in range(chunk_count):

            # 開始と終了を決める
            start_row = i * chunk_size
            end_row = start_row + chunk_size

            # 範囲出力
            chunk_data = ddf[start_row:end_row]

            # 分割したファイルの名前を決定
            split_file = Path.joinpath(parent_dir, f"{file_path.stem}_{start_row}_{end_row}")

            # 分割ファイルの出力
            chunk_data.to_csv(split_file, index=False, encoding=args.Encoding)
            print(f"->{split_file}の作成が完了しました")
    
    print("CSVの分割が完了しました。")
    sys.exit()

if __name__=="__main__":
    main()

引数の受け取り

対象ファイルや分割数と文字コードを柔軟に決めたかったので引数として受け取ることにしました。

引数の受け取りにはsysモジュールのargsではなくargsparserという別のライブラリを使用しました。

引数ごとのデータ型やデフォルト値を指定したり、
実行時に[-h]オプションを指定することで必要な引数と説明をしてくれるといった
一気におしゃれなプログラムにしてくれる優れものです。

自分しか使うことがないのでおしゃれにする必要はありませんが、、、

import argparse

#引数の受け取り
parser = argparse.ArgumentParser(description="ここに全体の説明が入るよおおおおおおお")  # クラスの作成
parser.add_argument("Path", help="ここに引数の説明をいれるよおおおおおお") # 第一引数(必須)
parser.add_argument("-c", "--Chunk", type=int, default=1000,help="ここにもはいるよおおおおおおお")  # オプション引数->デフォルト1000
parser.add_argument("-e", "--Encoding", default="cp932")  # オプション引数->デフォルトcp932

args = parser.parse_args()

実行結果（オプション[-h]で実行）

usage: argsparse.py [-h] [-c CHUNK] [-e ENCODING] Path

ここに全体の説明が入るよおおおおおおお

positional arguments:
  Path                  ここに引数の説明をいれるよおおおおおお

options:
  -h, --help            show this help message and exit
  -c CHUNK, --Chunk CHUNK
                        ここにもはいるよおおおおおおお
  -e ENCODING, --Encoding ENCODING

parser.add_argumentで引数を追加することができます。
普段使えそうなのものを抜粋するとこんな感じです。

"Path" # 必須位置引数
"-p" #  省略形
"--Path" # 必須ではないキーワード引数

help="説明を入れる感じ"
default="必須ではないときのデフォルト値を入れる"
required=True #引数を必須にする
type=int #データ型の指定(int)

ディレクトリ（ファイルチェック)

#ディレクトリ(ファイル)チェック
def checkPath(target_path):
    # 1.ディレクトリであれば配下のCSVファイルを検索しリストを返却
    # 2.CSVであればリストに格納して変換
    # 3.それ以外なら終了
    if target_path.is_dir():
        print("ディレクトリが見つかりました。")
        return target_path.glob('*.csv')
    elif target_path.is_file() and target_path.suffix == ".csv":
        print("CSVファイルが見つかりました。")
        return [target_path]
    else:
        print(f"{target_path}は存在しない、もしくはCSVファイルではない可能性があります。")
        sys.exit()

Pathlibを使用してディレクトリの確認とCSVファイルリストの作成を行いました。
ディレクトリを指定した場合はディレクトリ内のCSVファイルを検索(glob)しリスト返却します。
ファイルの場合は一つだけですがリストとして返却することで後の処理でループ処理ができるようにしました。

今後どこかで使うかもしれないと思ったのでディレクトリとファイルのどちらも実行できるようにしました。
ただ、めんどくさかったのでそれ以外は落ちるような処理です。

ディレクトリ作成

# ディレクトリ作成
def createDir(target_path):
    # ディレクトリが存在しなければ作成
    # ※なんとなく作っただけなのでメイン処理に埋め込んでもいい
    flg = False
    if not target_path.exists():
        target_path.mkdir()
        flg = True
    return flg

このぐらいならメイン処理に書いてもよかったのですがなんとなく関数として実装。
「なければ作る」よく見るアレ。

主処理

# 主処理
def main():

    # 引数の受け取り
    path = Path(args.Path)
    chunk_size = args.Chunk

    # CSVファイルリストの作成
    file_list = checkPath(path)

    # ファイルリストをループして処理
    for file_path in file_list:
        parent_dir = Path.joinpath(file_path.parent, file_path.stem)
        create_flg = createDir(parent_dir)

        # ディレクトリの作成ができたら出力->別に要らない処理
        if create_flg:
            print(f"ディレクトリを作成しました。->{parent_dir}")
        
        print(f"ファイル読み込み開始->{file_path}")

        # ファイル読み込み
        ddf = dd.read_csv(file_path, encoding=args.Encoding, header=0, dtype="object").compute()

        # 分割数の計算
        chunk_count = len(ddf) // chunk_size + 1

        # ループ処理で分割
        for i in range(chunk_count):

            # 開始と終了を決める
            start_row = i * chunk_size
            end_row = start_row + chunk_size

            # 範囲出力
            chunk_data = ddf[start_row:end_row]

            # 分割したファイルの名前を決定
            split_file = Path.joinpath(parent_dir, f"{file_path.stem}_{start_row}_{end_row}")

            # 分割ファイルの出力
            chunk_data.to_csv(split_file, index=False, encoding=args.Encoding)
            print(f"->{split_file}の作成が完了しました")
    
    print("CSVの分割が完了しました。")
    sys.exit()

if __name__=="__main__":
    main()

引数を受け取ってCSVを読み込むのですが、
今回はCSVの読み込みにpandasやcsvではなくdaskというものを使ってみました。
データサイエンティストの人たちの間では当たり前だと思うのですが、大量データを利用する場合はpandasよりdaskのほうが早いらしいです。

速度比較もしていないしよくわからないのですが「ほえええそうなんだ」と思いながら使いました。
pip installがややこしかった。
※ちなみにこの使い方が正しいのかはよくわかっていないがとりあえず使えたのでヨシ（現場猫）
気になる人がいたらpandasを使用してください。

また、分割に関してですが開始と終了位置を変えながら読み込んだ内容を出力することで分割を行っています。
ファイル名は重複しないように開始と終了を付与しています。

実行結果

作成できたのでCSVファイルに対して実行してみます。
CSVは以下のサイトからSampleを持ってきました。

image.png

いい感じの量ですね。
試しに5000行ずつ分割してみます。

python main.py annual-enterprise-survey-2021-financial-year-provisional-csv.csv -c 5000

結果

python main.py annual-enterprise-survey-2021-financial-year-provisional-csv.csv -c 5000
CSVファイルが見つかりました。
ディレクトリを作成しました。->annual-enterprise-survey-2021-financial-year-provisional-csv
ファイル読み込み開始->annual-enterprise-survey-2021-financial-year-provisional-csv.csv
->annual-enterprise-survey-2021-financial-year-provisional-csv/annual-enterprise-survey-2021-financial-year-provisional-csv_0_5000の作成が完了しました
->annual-enterprise-survey-2021-financial-year-provisional-csv/annual-enterprise-survey-2021-financial-year-provisional-csv_5000_10000の作成が完了しました
->annual-enterprise-survey-2021-financial-year-provisional-csv/annual-enterprise-survey-2021-financial-year-provisional-csv_10000_15000の作成が完了しました
->annual-enterprise-survey-2021-financial-year-provisional-csv/annual-enterprise-survey-2021-financial-year-provisional-csv_15000_20000の作成が完了しました
->annual-enterprise-survey-2021-financial-year-provisional-csv/annual-enterprise-survey-2021-financial-year-provisional-csv_20000_25000の作成が完了しました
->annual-enterprise-survey-2021-financial-year-provisional-csv/annual-enterprise-survey-2021-financial-year-provisional-csv_25000_30000の作成が完了しました
->annual-enterprise-survey-2021-financial-year-provisional-csv/annual-enterprise-survey-2021-financial-year-provisional-csv_30000_35000の作成が完了しました
->annual-enterprise-survey-2021-financial-year-provisional-csv/annual-enterprise-survey-2021-financial-year-provisional-csv_35000_40000の作成が完了しました
->annual-enterprise-survey-2021-financial-year-provisional-csv/annual-enterprise-survey-2021-financial-year-provisional-csv_40000_45000の作成が完了しました
CSVの分割が完了しました。

image.png

CSVの配下にディレクトリが作成され、分割されたファイルが格納されていることを確認しました。

終わりに

個人的には引数の受け取りやdaskなど初めて使用するライブラリの勉強にもなり(?)楽しかったです。
今後は出力のタイミングで分割していただけるように交渉したいので、
このプログラムの活躍する日が来ないことを願っています。

以上

2
2
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
2
2