parquet-converter

Parquet コンバーター

ビジネス上の課題

問題

データの保存と処理に関する課題：

• 大容量の CSV ファイルは処理に時間がかかります
• 型付きデータを効率的に保存できません
• 列指向クエリの処理に時間がかかります
• 最新のデータプラットフォームと互換性がありません

解決策

建設データを Parquet 形式に変換し、効率的な列指向ストレージ、高速なクエリ、データレイクハウスとの互換性を実現します。

技術的な実装

import pandas as pd
from typing import Dict, Any, List, Optional, Union
from dataclasses import dataclass, field
from datetime import datetime
from pathlib import Path
import json


class CompressionType:
    SNAPPY = "snappy"
    GZIP = "gzip"
    BROTLI = "brotli"
    ZSTD = "zstd"
    NONE = None


@dataclass
class ParquetSchema:
    columns: Dict[str, str]  # column_name: dtype
    partitions: List[str] = field(default_factory=list)
    row_group_size: int = 100000


@dataclass
class ConversionResult:
    source_path: str
    output_path: str
    source_format: str
    rows: int
    columns: int
    original_size_mb: float
    parquet_size_mb: float
    compression_ratio: float
    duration_seconds: float


class ParquetConverter:
    """Convert construction data to/from Parquet format."""

    def __init__(self, project_name: str = "Data Conversion"):
        self.project_name = project_name
        self.conversions: List[ConversionResult] = []
        self.schemas: Dict[str, ParquetSchema] = {}
        self._define_standard_schemas()

    def _define_standard_schemas(self):
        """Define standard schemas for construction data."""

        self.schemas['projects'] = ParquetSchema(
            columns={
                'project_id': 'string',
                'name': 'string',
                'project_type': 'category',
                'status': 'category',
                'start_date': 'datetime64[ns]',
                'end_date': 'datetime64[ns]',
                'budget': 'float64',
                'actual_cost': 'float64',
                'size_sf': 'float64',
                'location': 'string'
            },
            partitions=['project_type', 'status']
        )

        self.schemas['costs'] = ParquetSchema(
            columns={
                'transaction_id': 'string',
                'project_id': 'string',
                'cost_code': 'category',
                'description': 'string',
                'amount': 'float64',
                'transaction_date': 'datetime64[ns]',
                'vendor': 'string',
                'invoice_number': 'string'
            },
            partitions=['project_id']
        )

        self.schemas['schedule'] = ParquetSchema(
            columns={
                'activity_id': 'string',
                'project_id': 'string',
                'name': 'string',
                'wbs_code': 'string',
                'start_date': 'datetime64[ns]',
                'end_date': 'datetime64[ns]',
                'duration': 'int32',
                'progress': 'float32',
                'status': 'category'
            },
            partitions=['project_id']
        )

        self.schemas['qto'] = ParquetSchema(
            columns={
                'element_id': 'string',
                'project_id': 'string',
                'element_type': 'category',
                'name': 'string',
                'quantity': 'float64',
                'unit': 'category',
                'level': 'string',
                'material': 'string'
            },
            partitions=['project_id', 'element_type']
        )

    def add_schema(self, name: str, schema: ParquetSchema):
        """Add custom schema."""
        self.schemas[name] = schema

    def csv_to_parquet(self, csv_path: str, parquet_path: str,
                       schema_name: str = None,
                       compression: str = CompressionType.SNAPPY,
                       partition_cols: List[str] = None) -> ConversionResult:
        """Convert CSV to Parquet."""

        start_time = datetime.now()

        # Read CSV
        df = pd.read_csv(csv_path)

        # Apply schema if provided
        if schema_name and schema_name in self.schemas:
            schema = self.schemas[schema_name]
            df = self._apply_schema(df, schema)
            partition_cols = partition_cols or schema.partitions

        # Get original file size
        original_size = Path(csv_path).stat().st_size / (1024 * 1024)

        # Write Parquet
        if partition_cols:
            # Partitioned write
            available_partitions = [c for c in partition_cols if c in df.columns]
            if available_partitions:
                df.to_parquet(
                    parquet_path,
                    engine='pyarrow',
                    compression=compression,
                    partition_cols=available_partitions,
                    index=False
                )
            else:
                df.to_parquet(parquet_path, engine='pyarrow',
                             compression=compression, index=False)
        else:
            df.to_parquet(parquet_path, engine='pyarrow',
                         compression=compression, index=False)

        # Calculate parquet size
        if Path(parquet_path).is_dir():
            parquet_size = sum(f.stat().st_size for f in Path(parquet_path).rglob('*.parquet')) / (1024 * 1024)
        else:
            parquet_size = Path(parquet_path).stat().st_size / (1024 * 1024)

        duration = (datetime.now() - start_time).total_seconds()

        result = ConversionResult(
            source_path=csv_path,
            output_path=parquet_path,
            source_format='csv',
            rows=len(df),
            columns=len(df.columns),
            original_size_mb=round(original_size, 2),
            parquet_size_mb=round(parquet_size, 2),
            compression_ratio=round(original_size / parquet_size, 2) if parquet_size > 0 else 0,
            duration_seconds=round(duration, 2)
        )

        self.conversions.append(result)
        return result

    def excel_to_parquet(self, excel_path: str, parquet_path: str,
                         sheet_name: Union[str, int] = 0,
                         schema_name: str = None,
                         compression: str = CompressionType.SNAPPY) -> ConversionResult:
        """Convert Excel to Parquet."""

        start_time = datetime.now()

        # Read Excel
        df = pd.read_excel(excel_path, sheet_name=sheet_name)

        # Apply schema
        if schema_name and schema_name in self.schemas:
            df = self._apply_schema(df, self.schemas[schema_name])

        original_size = Path(excel_path).stat().st_size / (1024 * 1024)

        # Write Parquet
        df.to_parquet(parquet_path, engine='pyarrow',
                     compression=compression, index=False)

        parquet_size = Path(parquet_path).stat().st_size / (1024 * 1024)
        duration = (datetime.now() - start_time).total_seconds()

        result = ConversionResult(
            source_path=excel_path,
            output_path=parquet_path,
            source_format='excel',
            rows=len(df),
            columns=len(df.columns),
            original_size_mb=round(original_size, 2),
            parquet_size_mb=round(parquet_size, 2),
            compression_ratio=round(original_size / parquet_size, 2) if parquet_size > 0 else 0,
            duration_seconds=round(duration, 2)
        )

        self.conversions.append(result)
        return result

    def json_to_parquet(self, json_path: str, parquet_path: str,
                        schema_name: str = None,
                        compression: str = CompressionType.SNAPPY) -> ConversionResult:
        """Convert JSON to Parquet."""

        start_time = datetime.now()

        # Read JSON
        df = pd.read_json(json_path)

        if schema_name and schema_name in self.schemas:
            df = self._apply_schema(df, self.schemas[schema_name])

        original_size = Path(json_path).stat().st_size / (1024 * 1024)

        df.to_parquet(parquet_path, engine='pyarrow',
                     compression=compression, index=False)

        parquet_size = Path(parquet_path).stat().st_size / (1024 * 1024)
        duration = (datetime.now() - start_time).total_seconds()

        result = ConversionResult(
            source_path=json_path,
            output_path=parquet_path,
            source_format='json',
            rows=len(df),
            columns=len(df.columns),
            original_size_mb=round(original_size, 2),
            parquet_size_mb=round(parquet_size, 2),
            compression_ratio=round(original_size / parquet_size, 2) if parquet_size > 0 else 0,
            duration_seconds=round(duration, 2)
        )

        self.conversions.append(result)
        return result

    def parquet_to_csv(self, parquet_path: str, csv_path: str) -> ConversionResult:
        """Convert Parquet to CSV."""

        start_time = datetime.now()

        df = pd.read_parquet(parquet_path)

        if Path(parquet_path).is_dir():
            original_size = sum(f.stat().st_size for f in Path(parquet_path).rglob('*.parquet')) / (1024 * 1024)
        else:
            original_size = Path(parquet_path).stat().st_size / (1024 * 1024)

        df.to_csv(csv_path, index=False)

        csv_size = Path(csv_path).stat().st_size / (1024 * 1024)
        duration = (datetime.now() - start_time).total_seconds()

        result = ConversionResult(
            source_path=parquet_path,
            output_path=csv_path,
            source_format='parquet',
            rows=len(df),
            columns=len(df.columns),
            original_size_mb=round(original_size, 2),
            parquet_size_mb=round(csv_size, 2),  # Actually CSV size
            compression_ratio=round(csv_size / original_size, 2) if original_size > 0 else 0,
            duration_seconds=round(duration, 2)
        )

        self.conversions.append(result)
        return result

    def _apply_schema(self, df: pd.DataFrame, schema: ParquetSchema) -> pd.DataFrame:
        """Apply schema to DataFrame."""

        for col, dtype in schema.columns.items():
            if col in df.columns:
                try:
                    if dtype == 'category':
                        df[col] = df[col].astype('category')
                    elif dtype.startswith('datetime'):
                        df[col] = pd.to_datetime(df[col])
                    else:
                        df[col] = df[col].astype(dtype)
                except (ValueError, TypeError):
                    pass  # Keep original type if conversion fails

        return df

    def get_parquet_info(self, parquet_path: str) -> Dict[str, Any]:
        """Get information about a Parquet file."""

        import pyarrow.parquet as pq

        if Path(parquet_path).is_dir():
            # Partitioned dataset
            files = list(Path(parquet_path).rglob('*.parquet'))
            total_size = sum(f.stat().st_size for f in files) / (1024 * 1024)

            if files:
                sample = pq.read_table(str(files[0]))
                schema = sample.schema
            else:
                return {'error': 'No parquet files found'}

            return {
                'path': parquet_path,
                'type': 'partitioned',
                'num_files': len(files),
                'total_size_mb': round(total_size, 2),
                'columns': [f.name for f in schema],
                'dtypes': {f.name: str(f.type) for f in schema}
            }
        else:
            # Single file
            pf = pq.ParquetFile(parquet_path)
            metadata = pf.metadata

            return {
                'path': parquet_path,
                'type': 'single_file',
                'size_mb': round(Path(parquet_path).stat().st_size / (1024 * 1024), 2),
                'num_rows': metadata.num_rows,
                'num_columns': metadata.num_columns,
                'num_row_groups': metadata.num_row_groups,
                'columns': [pf.schema_arrow.field(i).name for i in range(metadata.num_columns)],
                'created_by': metadata.created_by
            }

    def query_parquet(self, parquet_path: str, columns: List[str] = None,
                      filters: List[tuple] = None) -> pd.DataFrame:
        """Query Parquet file with column selection and filtering."""

        return pd.read_parquet(
            parquet_path,
            columns=columns,
            filters=filters
        )

    def merge_parquet_files(self, input_paths: List[str],
                            output_path: str,
                            compression: str = CompressionType.SNAPPY) -> ConversionResult:
        """Merge multiple Parquet files into one."""

        start_time = datetime.now()

        dfs = [pd.read_parquet(p) for p in input_paths]
        merged = pd.concat(dfs, ignore_index=True)

        original_size = sum(Path(p).stat().st_size for p in input_paths) / (1024 * 1024)

        merged.to_parquet(output_path, engine='pyarrow',
                         compression=compression, index=False)

        parquet_size = Path(output_path).stat().st_size / (1024 * 1024)
        duration = (datetime.now() - start_time).total_seconds()

        return ConversionResult(
            source_path=str(input_paths),
            output_path=output_path,
            source_format='parquet_merge',
            rows=len(merged),
            columns=len(merged.columns),
            original_size_mb=round(original_size, 2),
            parquet_size_mb=round(parquet_size, 2),
            compression_ratio=round(original_size / parquet_size, 2) if parquet_size > 0 else 0,
            duration_seconds=round(duration, 2)
        )

    def get_conversion_summary(self) -> Dict[str, Any]:
        """Get summary of all conversions."""

        if not self.conversions:
            return {'total_conversions': 0}

        return {
            'total_conversions': len(self.conversions),
            'total_rows_processed': sum(c.rows for c in self.conversions),
            'original_size_mb': sum(c.original_size_mb for c in self.conversions),
            'parquet_size_mb': sum(c.parquet_size_mb for c in self.conversions),
            'avg_compression_ratio': round(
                sum(c.compression_ratio for c in self.conversions) / len(self.conversions), 2
            ),
            'total_duration_seconds': sum(c.duration_seconds for c in self.conversions)
        }

    def export_conversion_log(self, output_path: str) -> str:
        """Export conversion log to Excel."""

        with pd.ExcelWriter(output_path, engine='openpyxl') as writer:
            # Summary
            summary = self.get_conversion_summary()
            summary_df = pd.DataFrame([summary])
            summary_df.to_excel(writer, sheet_name='Summary', index=False)

            # Detailed log
            log_df = pd.DataFrame([{
                'Source': c.source_path,
                'Output': c.output_path,
                'Format': c.source_format,
                'Rows': c.rows,
                'Columns': c.columns,
                'Original Size (MB)': c.original_size_mb,
                'Parquet Size (MB)': c.parquet_size_mb,
                'Compression Ratio': c.compression_ratio,
                'Duration (s)': c.duration_seconds
            } for c in self.conversions])
            log_df.to_excel(writer, sheet_name='Conversions', index=False)

        return output_path

クイックスタート

# Create converter
converter = ParquetConverter("Project Data Migration")

# Convert CSV to Parquet
result = converter.csv_to_parquet(
    "costs.csv",
    "costs.parquet",
    schema_name="costs",
    compression="snappy"
)

print(f"Converted {result.rows} rows")
print(f"Compression ratio: {result.compression_ratio}x")
print(f"Size: {result.original_size_mb}MB -> {result.parquet_size_mb}MB")

よくあるユースケース

1. Excel から Parquet への変換

result = converter.excel_to_parquet(
    "project_data.xlsx",
    "project_data.parquet",
    schema_name="projects"
)

2. Parquet のクエリ

# Select specific columns with filter
df = converter.query_parquet(
    "costs.parquet",
    columns=['project_id', 'amount', 'transaction_date'],
    filters=[('amount', '>', 10000)]
)

3. ファイル情報の取得

info = converter.get_parquet_info("costs.parquet")
print(f"Rows: {info['num_rows']}")
print(f"Columns: {info['columns']}")

4. ファイルの結合

result = converter.merge_parquet_files(
    ["costs_2023.parquet", "costs_2024.parquet"],
    "costs_all.parquet"
)

リソース

• DDC Book：第 4.4 章 - 最新のデータ技術
• Apache Parquet：https://parquet.apache.org/
• PyArrow：https://arrow.apache.org/docs/python/
• ウェブサイト：https://datadrivenconstruction.io