【初心者歓迎】Python でParquet⇔CSV相互変換ツールを作ろう

       
import os
import sys
sys.path.append(os.path.join(os.path. dirname(__file__), '../libs'))
from create_ui import CreateUI
from parquet_util import ParquetUtil
from file_util import FileUtil

# ボタンが押された時の処理
def func(param):
    input_folder = param["input_folder"]
    output_folder = param["output_folder"]
    select = param["select_group"]
    csv_wcard = param["_csv_wildcard_entry"]
    parq_wildcard = param["_parq_wildcard_entry"]

    if select == 1 :
        wildcard = csv_wcard
        target_ext = ".parquet"
    else:
        wildcard = parq_wildcard
        target_ext = ".csv"

    fu = FileUtil()
    files = fu.get_files(input_folder,wildcard=wildcard) 
    for file in files:
        _,name = os.path.split(file)
        input_file = os.path.join(input_folder,name)
        output_file = os.path.join(output_folder,fu.change_extension(name,target_ext))       
        ParquetUtil.convert(input_file,output_file)
   

# インスタンス生成
ui = CreateUI(func)


# 画面の表示
ui.show()

import ParquetUtil

# CSVをParquetに変換
ParquetUtil.csv_to_parq("data.csv", "data.parquet")

# Parquetファイルの先頭5行を表示
ParquetUtil.head("data.parquet", 5)

import pandas as pd
import os

class ParquetUtil:
    @staticmethod
    def csv_to_parq(csv_file_path, parq_file_path):
        """
        CSVファイルをParquetファイルに変換する。

        パラメータ:
        csv_file_path (str): 入力CSVファイルのパス。
        parq_file_path (str): 出力Parquetファイルのパス。
        """
        df = pd.read_csv(csv_file_path)
        df.to_parquet(parq_file_path, index=False)
        print(f"{csv_file_path} を {parq_file_path} に変換しました")

    @staticmethod
    def parq_to_csv(parq_file_path, csv_file_path):
        """
        ParquetファイルをCSVファイルに変換する。

        パラメータ:
        parq_file_path (str): 入力Parquetファイルのパス。
        csv_file_path (str): 出力CSVファイルのパス。
        """
        df = pd.read_parquet(parq_file_path)
        df.to_csv(csv_file_path, index=False)
        print(f"{parq_file_path} を {csv_file_path} に変換しました")

    @staticmethod
    def head(parq_file_path, num_rows=5):
        """
        Parquetファイルの先頭を表示する。

        パラメータ:
        parq_file_path (str): Parquetファイルのパス。
        num_rows (int): 表示する行数。デフォルトは5。
        """
        df = pd.read_parquet(parq_file_path)
        print(df.head(num_rows))

    @staticmethod
    def tail(parq_file_path, num_rows=5):
        """
        Parquetファイルの末尾を表示する。

        パラメータ:
        parq_file_path (str): Parquetファイルのパス。
        num_rows (int): 表示する行数。デフォルトは5。
        """
        df = pd.read_parquet(parq_file_path)
        print(df.tail(num_rows))

    @staticmethod
    def sample(parq_file_path, start_row, num_rows):
        """
        Parquetファイルの任意の位置から指定した行数を表示する。

        パラメータ:
        parq_file_path (str): Parquetファイルのパス。
        start_row (int): 開始行のインデックス。
        num_rows (int): 表示する行数。
        """
        df = pd.read_parquet(parq_file_path)
        print(df.iloc[start_row:start_row + num_rows])

    @staticmethod
    def info(parq_file_path):
        """
        Parquetファイルの情報を表示する。

        パラメータ:
        parq_file_path (str): Parquetファイルのパス。
        """
        df = pd.read_parquet(parq_file_path)
        print(f"形状: {df.shape}")
        print(f"列: {df.columns.tolist()}")
        print(f"データ型: {df.dtypes}")

    @staticmethod
    def convert_file(input_file_path, output_file_path):
        """
        ファイルの拡張子を判定して自動的に変換する。

        パラメータ:
        input_file_path (str): 入力ファイル（CSVまたはParquet）のパス。
        output_file_path (str): 出力ファイル（ParquetまたはCSV）のパス。
        """
        input_ext = os.path.splitext(input_file_path)[1].lower()
        output_ext = os.path.splitext(output_file_path)[1].lower()

        if input_ext == '.csv' and output_ext == '.parquet':
            ParquetUtil.csv_to_parq(input_file_path, output_file_path)
        elif input_ext == '.parquet' and output_ext == '.csv':
            ParquetUtil.parq_to_csv(input_file_path, output_file_path)
        else:
            print("サポートされていないファイル変換です。正しい入力および出力ファイル形式を指定してください。")

    @staticmethod
    def convert(input_path, output_path):
        """
        ファイルまたはフォルダ内のすべてのファイルの拡張子を判定して自動的に変換する。

        パラメータ:
        input_path (str): 入力ファイルまたはフォルダのパス。
        output_path (str): 出力ファイルまたはフォルダのパス。
        """
        if os.path.isfile(input_path):
            # ファイルの変換
            ParquetUtil.convert_file(input_path, output_path)
        elif os.path.isdir(input_path):
            # フォルダ内のすべてのファイルの変換
            if not os.path.exists(output_path):
                os.makedirs(output_path)
            for filename in os.listdir(input_path):
                input_file = os.path.join(input_path, filename)
                output_file = os.path.join(output_path, os.path.splitext(filename)[0] + ('.parquet' if filename.lower().endswith('.csv') else '.csv'))
                ParquetUtil.convert_file(input_file, output_file)
                print(f"{input_file}->{output_file}")
        else:
            print("指定されたパスはファイルでもフォルダでもありません。")

title:CSV⇔Parquet相互変換ツール
mode:light
max:1024,350
size:500,350
min:500,350
input_folder:入力元フォルダ,フォルダ選択
output_folder:出力先フォルダ,フォルダ選択
select_group:
{
    p2c_exec_radio:Parquet ⇒ CSV変換,select_group,1
    label:ワイルドカード
    _parq_wildcard_entry:150,*.parquet
}
{
    c2p_exec_radio:CSV ⇒ Parquet変換,select_group,2
    label:ワイルドカード
    _csv_wildcard_entry:150,*.csv
}
exec_button:ファイル変換