Parquet Converter

Business Case

Problem Statement

Data storage and processing challenges:

• Large CSV files are slow to process
• Inefficient storage of typed data
• Column-oriented queries are slow
• Incompatible with modern data platforms

Solution

Convert construction data to Parquet format for efficient columnar storage, faster queries, and compatibility with data lakehouses.

Technical Implementation

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_p