Big Data Analysis

Business Case

Problem Statement

Large-scale data analysis challenges:

• Processing millions of records

• Cross-project benchmarking

• Pattern recognition at scale

• Memory and performance constraints

Solution

Scalable big data analysis framework for construction data using efficient data structures and parallel processing patterns.

Technical Implementation

import pandas as pd from typing import Dict, Any, List, Optional, Callable, Iterator from dataclasses import dataclass, field from datetime import datetime, date from enum import Enum import json

class AnalysisType(Enum): BENCHMARK = "benchmark" TREND = "trend" ANOMALY = "anomaly" CORRELATION = "correlation" CLUSTERING = "clustering" AGGREGATION = "aggregation"

class MetricType(Enum): COST_PER_SF = "cost_per_sf" DURATION_PER_SF = "duration_per_sf" PRODUCTIVITY = "productivity" CHANGE_ORDER_RATE = "change_order_rate" SAFETY_RATE = "safety_rate" QUALITY_SCORE = "quality_score"

@dataclass class ProjectRecord: project_id: str name: str project_type: str location: str size_sf: float duration_days: int total_cost: float start_date: date metrics: Dict[str, float] = field(default_factory=dict) attributes: Dict[str, Any] = field(default_factory=dict)

@dataclass class BenchmarkResult: metric: str mean: float median: float std: float min_val: float max_val: float percentile_25: float percentile_75: float sample_size: int

class BigDataAnalyzer: """Analyze large-scale construction datasets."""

def __init__(self, name: str = "Construction Analytics"):
    self.name = name
    self.projects: List[ProjectRecord] = []
    self.df: Optional[pd.DataFrame] = None
    self.benchmarks: Dict[str, BenchmarkResult] = {}

def load_from_dataframe(self, df: pd.DataFrame):
    """Load project data from DataFrame."""

    self.df = df.copy()
    self.projects = []

    for _, row in df.iterrows():
        project = ProjectRecord(
            project_id=str(row.get('project_id', '')),
            name=str(row.get('name', '')),
            project_type=str(row.get('project_type', '')),
            location=str(row.get('location', '')),
            size_sf=float(row.get('size_sf', 0)),
            duration_days=int(row.get('duration_days', 0)),
            total_cost=float(row.get('total_cost', 0)),
            start_date=pd.to_datetime(row.get('start_date')).date() if pd.notna(row.get('start_date')) else date.today()
        )
        # Add calculated metrics
        if project.size_sf > 0:
            project.metrics['cost_per_sf'] = project.total_cost / project.size_sf
            project.metrics['duration_per_1000sf'] = project.duration_days / (project.size_sf / 1000)

        self.projects.append(project)

def load_from_parquet(self, path: str):
    """Load data from Parquet file."""
    df = pd.read_parquet(path)
    self.load_from_dataframe(df)

def stream_process(self, file_path: str, chunk_size: int = 10000,
                   processor: Callable = None) -> Iterator[Dict[str, Any]]:
    """Process large file in chunks."""

    for chunk in pd.read_csv(file_path, chunksize=chunk_size):
        if processor:
            result = processor(chunk)
            yield result
        else:
            yield {'rows': len(chunk), 'columns': list(chunk.columns)}

def calculate_benchmarks(self, metric_column: str,
                         group_by: str = None) -> Dict[str, BenchmarkResult]:
    """Calculate benchmarks for a metric."""

    if self.df is None or self.df.empty:
        return {}

    results = {}

    if group_by and group_by in self.df.columns:
        groups = self.df.groupby(group_by)
        for group_name, group_df in groups:
            values = group_df[metric_column].dropna()
            if len(values) > 0:
                results[str(group_name)] = self._calculate_stats(values, metric_column)
    else:
        values = self.df[metric_column].dropna()
        if len(values) > 0:
            results['all'] = self._calculate_stats(values, metric_column)

    self.benchmarks.update(results)
    return results

def _calculate_stats(self, values: pd.Series, metric: str) -> BenchmarkResult:
    """Calculate statistics for a series."""

    return BenchmarkResult(
        metric=metric,
        mean=round(values.mean(), 2),
        median=round(values.median(), 2),
        std=round(values.std(), 2),
        min_val=round(values.min(), 2),
        max_val=round(values.max(), 2),
        percentile_25=round(values.quantile(0.25), 2),
        percentile_75=round(values.quantile(0.75), 2),
        sample_size=len(values)
    )

def find_anomalies(self, metric_column: str,
                   threshold_std: float = 2.0) -> pd.DataFrame:
    """Find anomalies based on standard deviation threshold."""

    if self.df is None or self.df.empty:
        return pd.DataFrame()

    values = self.df[metric_column]
    mean = values.mean()
    std = values.std()

    lower = mean - (threshold_std * std)
    upper = mean + (threshold_std * std)

    anomalies = self.df[(values < lower) | (values > upper)].copy()
    anomalies['anomaly_type'] = anomalies[metric_column].apply(
        lambda x: 'high' if x > upper else 'low'
    )
    anomalies['deviation'] = ((anomalies[metric_column] - mean) / std).round(2)

    return anomalies

def analyze_trends(self, metric_column: str,
                   date_column: str,
                   period: str = 'Y') -> pd.DataFrame:
    """Analyze trends over time."""

    if self.df is None or self.df.empty:
        return pd.DataFrame()

    df = self.df.copy()
    df[date_column] = pd.to_datetime(df[date_column])
    df['period'] = df[date_column].dt.to_period(period)

    trends = df.groupby('period').agg({
        metric_column: ['mean', 'median', 'count', 'std']
    }).round(2)

    trends.columns = ['mean', 'median', 'count', 'std']
    trends = trends.reset_index()
    trends['period'] = trends['period'].astype(str)

    # Calculate year-over-year change
    trends['yoy_change'] = trends['mean'].pct_change().round(4) * 100

    return trends

def calculate_correlations(self, columns: List[str]) -> pd.DataFrame:
    """Calculate correlations between metrics."""

    if self.df is None or self.df.empty:
        return pd.DataFrame()

    available_cols = [c for c in columns if c in self.df.columns]
    return self.df[available_cols].corr().round(3)

def segment_analysis(self, metric_column: str,
                     segment_column: str) -> pd.DataFrame:
    """Analyze metric by segments."""

    if self.df is None or self.df.empty:
        return pd.DataFrame()

    results = self.df.groupby(segment_column).agg({
        metric_column: ['count', 'mean', 'median', 'std', 'min', 'max']
    }).round(2)

    results.columns = ['count', 'mean', 'median', 'std', 'min', 'max']
    results = results.reset_index()

    # Calculate percentage of total
    total_count = results['count'].sum()
    results['pct_of_total'] = (results['count'] / total_count * 100).round(1)

    return results.sort_values('count', ascending=False)

def percentile_rank(self, project_id: str, metric_column: str) -> Dict[str, Any]:
    """Get percentile rank for a specific project."""

    if self.df is None or self.df.empty:
        return {}

    project = self.df[self.df['project_id'] == project_id]
    if project.empty:
        return {'error': 'Project not found'}

    value = project[metric_column].values[0]
    all_values = self.df[metric_column].dropna()

    percentile = (all_values < value).sum() / len(all_values) * 100

    benchmark = self.benchmarks.get('all') or self._calculate_stats(all_values, metric_column)

    return {
        'project_id': project_id,
        'metric': metric_column,
        'value': round(value, 2),
        'percentile': round(percentile, 1),
        'comparison': {
            'mean': benchmark.mean,
            'median': benchmark.median,
            'vs_mean': round((value / benchmark.mean - 1) * 100, 1),
            'vs_median': round((value / benchmark.median - 1) * 100, 1)
        }
    }

def generate_summary_stats(self) -> Dict[str, Any]:
    """Generate summary statistics for the dataset."""

    if self.df is None or self.df.empty:
        return {}

    numeric_cols = self.df.select_dtypes(include=['number']).columns

    return {
        'total_projects': len(self.df),
        'date_range': {
            'min': str(self.df['start_date'].min()) if 'start_date' in self.df.columns else None,
            'max': str(self.df['start_date'].max()) if 'start_date' in self.df.columns else None
        },
        'project_types': self.df['project_type'].nunique() if 'project_type' in self.df.columns else 0,
        'locations': self.df['location'].nunique() if 'location' in self.df.columns else 0,
        'total_value': self.df['total_cost'].sum() if 'total_cost' in self.df.columns else 0,
        'total_sf': self.df['size_sf'].sum() if 'size_sf' in self.df.columns else 0,
        'numeric_columns': list(numeric_cols)
    }

def export_analysis(self, output_path: str,
                    metrics: List[str] = None) -> str:
    """Export analysis results to Excel."""

    metrics = metrics or ['cost_per_sf', 'duration_days']

    with pd.ExcelWriter(output_path, engine='openpyxl') as writer:
        # Summary
        summary = self.generate_summary_stats()
        summary_df = pd.DataFrame([{
            'Total Projects': summary.get('total_projects', 0),
            'Project Types': summary.get('project_types', 0),
            'Locations': summary.get('locations', 0),
            'Total Value ($)': summary.get('total_value', 0),
            'Total SF': summary.get('total_sf', 0)
        }])
        summary_df.to_excel(writer, sheet_name='Summary', index=False)

        # Benchmarks
        for metric in metrics:
            if metric in self.df.columns:
                benchmarks = self.calculate_benchmarks(metric, 'project_type')
                if benchmarks:
                    bench_data = [{
                        'Segment': k,
                        'Mean': v.mean,
                        'Median': v.median,
                        'Std': v.std,
                        'Min': v.min_val,
                        'Max': v.max_val,
                        'P25': v.percentile_25,
                        'P75': v.percentile_75,
                        'Count': v.sample_size
                    } for k, v in benchmarks.items()]
                    bench_df = pd.DataFrame(bench_data)
                    sheet_name = f"Benchmark_{metric}"[:31]
                    bench_df.to_excel(writer, sheet_name=sheet_name, index=False)

        # Anomalies
        for metric in metrics:
            if metric in self.df.columns:
                anomalies = self.find_anomalies(metric)
                if not anomalies.empty:
                    sheet_name = f"Anomalies_{metric}"[:31]
                    anomalies.to_excel(writer, sheet_name=sheet_name, index=False)

    return output_path

Quick Start

Create analyzer

analyzer = BigDataAnalyzer("Multi-Project Analysis")

Load data

df = pd.DataFrame([ {'project_id': 'P001', 'name': 'Office A', 'project_type': 'Office', 'location': 'NYC', 'size_sf': 50000, 'duration_days': 365, 'total_cost': 15000000, 'start_date': '2023-01-01'}, {'project_id': 'P002', 'name': 'Office B', 'project_type': 'Office', 'location': 'LA', 'size_sf': 75000, 'duration_days': 400, 'total_cost': 20000000, 'start_date': '2023-03-01'}, {'project_id': 'P003', 'name': 'Warehouse', 'project_type': 'Industrial', 'location': 'Chicago', 'size_sf': 100000, 'duration_days': 200, 'total_cost': 12000000, 'start_date': '2023-06-01'} ])

Add calculated metric

df['cost_per_sf'] = df['total_cost'] / df['size_sf']

analyzer.load_from_dataframe(df)

Calculate benchmarks

benchmarks = analyzer.calculate_benchmarks('cost_per_sf', 'project_type') for segment, stats in benchmarks.items(): print(f"{segment}: ${stats.mean:.2f}/SF (median: ${stats.median:.2f}/SF)")

Common Use Cases

1. Find Anomalies

anomalies = analyzer.find_anomalies('cost_per_sf', threshold_std=2.0) print(f"Found {len(anomalies)} anomalous projects")

2. Trend Analysis

trends = analyzer.analyze_trends('cost_per_sf', 'start_date', 'Y') print(trends)

3. Project Ranking

ranking = analyzer.percentile_rank('P001', 'cost_per_sf') print(f"Project is at {ranking['percentile']}th percentile")

4. Segment Analysis

segments = analyzer.segment_analysis('cost_per_sf', 'project_type') print(segments)

Resources

• DDC Book: Chapter 4.4 - Modern Data Technologies

• Apache Parquet: https://parquet.apache.org/

• Website: https://datadrivenconstruction.io