【C#】Parquetファイルを読み書きするには？Pythonと速度も比較！

using Parquet;
using Parquet.Data;

// Parquet ファイルを開く（読み取りモード）
using (var stream = File.OpenRead(@"d:\data.parquet"))
{
    // ParquetReader を作成（同期バージョン）
    using (var reader = ParquetReader.Open(stream))
    {
        // ファイル内のカラム情報（スキーマ）を取得
        var dataFields = reader.Schema.GetDataFields();

        // 最初の RowGroup を開く（0番目）
        var groupReader = reader.OpenRowGroupReader(0);

        // 各カラム（フィールド）に対して処理
        foreach (var field in dataFields)
        {
            // 該当カラムのデータを読み込む（同期）
            var column = groupReader.ReadColumn(field);

            // カラム名を表示
            Console.WriteLine($"Column: {field.Name}");

            // カラム内の各値を表示
            foreach (var val in column.Data)
            {
                Console.WriteLine(val);
            }
        }
    }
}

using System.IO;
using Parquet;
using Parquet.Data;
using Parquet.Schema;

async Task WriteParquetFile()
{
    // Parquetファイルの書き込み先パス
    using (var stream = System.IO.File.Create(@"d:\output.parquet"))
    {
        // Parquetのスキーマ（カラム名とデータ型）を定義
        var schema = new ParquetSchema(
            new DataField<int>("id"),
            new DataField<string>("name")
        );

        // id列のデータを定義（int型）
        var idColumn = new DataColumn(schema.GetDataFields()[0], new int[] { 1, 2, 3 });

        // name列のデータを定義（string型）
        var nameColumn = new DataColumn(schema.GetDataFields()[1], new string[] { "Alice", "Bob", "Carol" });

        // ParquetWriterを作成（非同期）
        using var writer = await ParquetWriter.CreateAsync(schema, stream);

        // 1つのRowGroup（レコードのまとまり）を作成
        using var rowGroup = writer.CreateRowGroup();

        // 各カラムのデータを書き込み
        await rowGroup.WriteColumnAsync(idColumn);
        await rowGroup.WriteColumnAsync(nameColumn);
    }
}

# Parquetファイルの読み込み
DataTable dt = await ParquetUtils.Read(@"D:\hoge.parquet");

# Parquet ファイルの書き込み
await ParquetUtils.Write(@"D:\hage.parquet", dt);

using System;
using System.Collections.Generic;
using System.Data; // DataTable を使うために必要
using System.IO;
using System.Linq;
using System.Threading.Tasks;
using System.Windows.Documents;
using Parquet;
using Parquet.Data; // DataTable への拡張メソッドのために必要 (Parquet.Net パッケージに含まれる)
using Parquet.Schema; // ParquetSchema のために必要

namespace ParquetDemo
{
    internal static class ParquetUtils
    {
        /// <summary>
        /// 指定されたParquetファイルを読み込み、その内容をDataTableとして返します。
        /// カラムの定義はParquetファイルから自動的に推論されます。
        /// </summary>
        /// <param name="filePath">読み込むParquetファイルのパス</param>
        /// <returns>Parquetファイルから読み込まれたデータを含むDataTable</returns>
        /// <exception cref="FileNotFoundException">指定されたファイルが見つからない場合</exception>
        /// <exception cref="ParquetException">Parquetファイルの読み込み中にエラーが発生した場合</exception>
        public static async Task<DataTable> Read(string filePath)
        {
            using (Stream fs = System.IO.File.OpenRead(filePath))
            {
                using (ParquetReader reader = await ParquetReader.CreateAsync(fs))
                {
                    // DataTable構築
                    var dt = new DataTable();
                    var dataFields = reader.Schema.GetDataFields();

                    foreach (var df in dataFields)
                    {
                        var clrType = Nullable.GetUnderlyingType(df.ClrNullableIfHasNullsType)
                                      ?? df.ClrNullableIfHasNullsType;
                        dt.Columns.Add(df.Name, clrType);
                    }

                    for (int i = 0; i < reader.RowGroupCount; i++)
                    {
                        using var rgReader = reader.OpenRowGroupReader(i);
                        var columns = new Parquet.Data.DataColumn[dataFields.Length];

                        for (int j = 0; j < dataFields.Length; j++)
                        {
                            columns[j] = await rgReader.ReadColumnAsync(dataFields[j]);
                        }

                        int rowCount = columns[0].Data.Length;
                        for (int r = 0; r < rowCount; r++)
                        {
                            var rowValues = new object[dataFields.Length];
                            for (int c = 0; c < dataFields.Length; c++)
                            {
                                rowValues[c] = columns[c].Data.GetValue(r)!;
                            }
                            dt.Rows.Add(rowValues);
                        }
                    }
                    return dt;
                }
            }
        }


        public static async Task Write(string filePath, DataTable dt)
        {
            using Stream output = File.Create(filePath);

            // スキーマ構築
            var fields = new List<DataField>();
            foreach (System.Data.DataColumn col in dt.Columns)
            {
                Type clrType = col.DataType;

                // Parquet.Net は null 許容型をサポートしているが、DataTable 側は非対応なのでそのまま
                if (clrType == typeof(int)) fields.Add(new DataField<int>(col.ColumnName));
                else if (clrType == typeof(long)) fields.Add(new DataField<long>(col.ColumnName));
                else if (clrType == typeof(double)) fields.Add(new DataField<double>(col.ColumnName));
                else if (clrType == typeof(bool)) fields.Add(new DataField<bool>(col.ColumnName));
                else if (clrType == typeof(string)) fields.Add(new DataField<string>(col.ColumnName));
                else if (clrType == typeof(DateTime)) fields.Add(new DataField<DateTime>(col.ColumnName));
                else
                    throw new NotSupportedException($"型 {clrType} は現在未対応です");
            }

            var schema = new ParquetSchema(fields);

            using var writer = await ParquetWriter.CreateAsync(schema, output);
            using var rowGroupWriter = writer.CreateRowGroup();

            foreach (DataField field in fields)
            {
                var col = dt.Columns[field.Name];
                Array values = Array.CreateInstance(field.ClrType, dt.Rows.Count);
                for (int i = 0; i < dt.Rows.Count; i++)
                {
                    object? raw = dt.Rows[i][col];
                    values.SetValue(raw == DBNull.Value ? null : raw, i);
                }

                var dataCol = new Parquet.Data.DataColumn(field, values);
                await rowGroupWriter.WriteColumnAsync(dataCol);
            }
        }
    }
}

import pandas as pd
import time

# 測定対象のファイルパス
input_path = r"P:\can_in.parquet"
output_path = r"P:\can_out.parquet"

# 読み込み時間計測
start_read = time.time()
df = pd.read_parquet(input_path, engine="pyarrow")
end_read = time.time()
print(f"読み込み時間: {end_read - start_read:.3f} 秒")

# 書き出し時間計測
start_write = time.time()
df.to_parquet(output_path, engine="pyarrow", compression="snappy")
end_write = time.time()
print(f"保存時間: {end_write - start_write:.3f} 秒")