big-data-analysis

これはレビュー時に保存された抜粋です。完全で最新の内容は外部ソースを確認してください。

---
name: "big-data-analysis"
description: "Analyze large-scale construction datasets. Process thousands of projects for patterns, benchmarks, and predictive insights."
homepage: "https://datadrivenconstruction.io"
metadata: {"openclaw": {"emoji": "🔢", "os": ["darwin", "linux", "win32"], "homepage": "https://datadrivenconstruction.io", "requires": {"bins": ["python3"]}}}
---
# 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

```python
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