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