data-profiler

建設データのプロファイリング

概要

建設データを分析し、その特性、分布、品質、パターンを把握します。データ品質の評価、ETL の計画、プロジェクトに影響が及ぶ前のデータ問題の特定に不可欠です。

ビジネスケース

建設データを使用する前に、以下を把握する必要があります。

• どのようなデータ型が含まれているか
• 値の分布
• 欠損データのパターン
• 異常値と外れ値
• 参照整合性の問題

このスキルはデータのプロファイリングを行い、これらの疑問に答えるとともに、実行可能な洞察を提供します。

技術的な実装

from dataclasses import dataclass, field
from typing import List, Dict, Any, Optional, Tuple
import pandas as pd
import numpy as np
from datetime import datetime
import json

@dataclass
class ColumnProfile:
    name: str
    data_type: str
    inferred_type: str  # More specific: project_id, cost, date, csi_code, etc.
    total_count: int
    null_count: int
    null_percentage: float
    unique_count: int
    uniqueness_ratio: float
    # For numeric columns
    min_value: Optional[float] = None
    max_value: Optional[float] = None
    mean_value: Optional[float] = None
    median_value: Optional[float] = None
    std_dev: Optional[float] = None
    # For string columns
    min_length: Optional[int] = None
    max_length: Optional[int] = None
    avg_length: Optional[float] = None
    # Top values
    top_values: List[Tuple[Any, int]] = field(default_factory=list)
    # Patterns
    common_patterns: List[str] = field(default_factory=list)
    # Quality flags
    quality_issues: List[str] = field(default_factory=list)

@dataclass
class DataProfile:
    source_name: str
    row_count: int
    column_count: int
    columns: List[ColumnProfile]
    duplicate_rows: int
    memory_usage: str
    profiled_at: datetime
    quality_score: float
    recommendations: List[str]

class ConstructionDataProfiler:
    """Profile construction data for quality and characteristics."""

    # Known construction data patterns
    CONSTRUCTION_PATTERNS = {
        'csi_code': r'^\d{2}\s?\d{2}\s?\d{2}$',
        'project_id': r'^[A-Z]{2,4}[-_]?\d{3,6}$',
        'cost_code': r'^\d{2}[-.]?\d{2,4}$',
        'wbs': r'^[\d.]+$',
        'phone': r'^\(?\d{3}\)?[-.\s]?\d{3}[-.\s]?\d{4}$',
        'email': r'^[\w.-]+@[\w.-]+\.\w+$',
        'date_iso': r'^\d{4}-\d{2}-\d{2}',
        'date_us': r'^\d{1,2}/\d{1,2}/\d{2,4}$',
        'currency': r'^\$?[\d,]+\.?\d{0,2}$',
        'percentage': r'^\d+\.?\d*%?$',
    }

    # Construction-specific column name patterns
    COLUMN_TYPE_HINTS = {
        'project': ['project_id', 'project_name', 'proj', 'job'],
        'cost': ['cost', 'amount', 'price', 'total', 'budget', 'actual'],
        'date': ['date', 'start', 'finish', 'end', 'created', 'modified'],
        'quantity': ['qty', 'quantity', 'count', 'units'],
        'csi': ['csi', 'division', 'masterformat', 'spec'],
        'location': ['location', 'area', 'zone', 'floor', 'level'],
        'person': ['owner', 'manager', 'superintendent', 'foreman', 'contact'],
    }

    def __init__(self):
        self.profiles: Dict[str, DataProfile] = {}

    def profile_dataframe(self, df: pd.DataFrame, source_name: str) -> DataProfile:
        """Profile a pandas DataFrame."""
        columns = []

        for col in df.columns:
            col_profile = self._profile_column(df[col], col)
            columns.append(col_profile)

        # Calculate duplicates
        duplicate_rows = len(df) - len(df.drop_duplicates())

        # Calculate memory usage
        memory_bytes = df.memory_usage(deep=True).sum()
        if memory_bytes < 1024:
            memory_usage = f"{memory_bytes} B"
        elif memory_bytes < 1024**2:
            memory_usage = f"{memory_bytes/1024:.1f} KB"
        else:
            memory_usage = f"{memory_bytes/1024**2:.1f} MB"

        # Calculate overall quality score
        quality_score = self._calculate_quality_score(columns)

        # Generate recommendations
        recommendations = self._generate_recommendations(columns, df)

        profile = DataProfile(
            source_name=source_name,
            row_count=len(df),
            column_count=len(df.columns),
            columns=columns,
            duplicate_rows=duplicate_rows,
            memory_usage=memory_usage,
            profiled_at=datetime.now(),
            quality_score=quality_score,
            recommendations=recommendations
        )

        self.profiles[source_name] = profile
        return profile

    def _profile_column(self, series: pd.Series, name: str) -> ColumnProfile:
        """Profile a single column."""
        total_count = len(series)
        null_count = series.isnull().sum()
        null_percentage = (null_count / total_count * 100) if total_count > 0 else 0

        # Get non-null values for analysis
        non_null = series.dropna()
        unique_count = non_null.nunique()
        uniqueness_ratio = unique_count / len(non_null) if len(non_null) > 0 else 0

        profile = ColumnProfile(
            name=name,
            data_type=str(series.dtype),
            inferred_type=self._infer_construction_type(series, name),
            total_count=total_count,
            null_count=null_count,
            null_percentage=round(null_percentage, 2),
            unique_count=unique_count,
            uniqueness_ratio=round(uniqueness_ratio, 4)
        )

        # Numeric analysis
        if pd.api.types.is_numeric_dtype(series):
            profile.min_value = float(non_null.min()) if len(non_null) > 0 else None
            profile.max_value = float(non_null.max()) if len(non_null) > 0 else None
            profile.mean_value = float(non_null.mean()) if len(non_null) > 0 else None
            profile.median_value = float(non_null.median()) if len(non_null) > 0 else None
            profile.std_dev = float(non_null.std()) if len(non_null) > 1 else None

            # Check for outliers
            if len(non_null) > 10 and profile.std_dev:
                outliers = non_null[abs(non_null - profile.mean_value) > 3 * profile.std_dev]
                if len(outliers) > 0:
                    profile.quality_issues.append(f"{len(outliers)} potential outliers detected")

            # Check for negative costs
            if any(hint in name.lower() for hint in ['cost', 'amount', 'price', 'total']):
                negatives = (non_null < 0).sum()
                if negatives > 0:
                    profile.quality_issues.append(f"{negatives} negative values in cost column")

        # String analysis
        elif pd.api.types.is_object_dtype(series) or pd.api.types.is_string_dtype(series):
            str_series = non_null.astype(str)
            lengths = str_series.str.len()
            profile.min_length = int(lengths.min()) if len(lengths) > 0 else None
            profile.max_length = int(lengths.max()) if len(lengths) > 0 else None
            profile.avg_length = float(lengths.mean()) if len(lengths) > 0 else None

            # Detect patterns
            profile.common_patterns = self._detect_patterns(str_series)

        # Top values
        if len(non_null) > 0:
            value_counts = non_null.value_counts().head(5)
            profile.top_values = list(zip(value_counts.index.tolist(), value_counts.values.tolist()))

        # Quality checks
        if null_percentage > 50:
            profile.quality_issues.append("High null rate (>50%)")
        if uniqueness_ratio == 1.0 and total_count > 100:
            profile.quality_issues.append("All unique values - possible ID column")
        if uniqueness_ratio < 0.01 and unique_count > 1:
            profile.quality_issues.append("Low cardinality - possible category")

        return profile

    def _infer_construction_type(self, series: pd.Series, name: str) -> str:
        """Infer construction-specific data type."""
        name_lower = name.lower()

        # Check column name hints
        for type_name, hints in self.COLUMN_TYPE_HINTS.items():
            if any(hint in name_lower for hint in hints):
                return type_name

        # Check data patterns
        non_null = series.dropna().astype(str)
        if len(non_null) == 0:
            return "unknown"

        sample = non_null.head(100)

        for pattern_name, pattern in self.CONSTRUCTION_PATTERNS.items():
            matches = sample.str.match(pattern, na=False).sum()
            if matches / len(sample) > 0.8:
                return pattern_name

        # Default to pandas dtype
        if pd.api.types.is_numeric_dtype(series):
            return "numeric"
        elif pd.api.types.is_datetime64_any_dtype(series):
            return "datetime"
        else:
            return "text"

    def _detect_patterns(self, str_series: pd.Series) -> List[str]:
        """Detect common patterns in string data."""
        patterns_found = []

        sample = str_series.head(1000)

        for pattern_name, pattern in self.CONSTRUCTION_PATTERNS.items():
            matches = sample.str.match(pattern, na=False).sum()
            if matches / len(sample) > 0.1:
                patterns_found.append(f"{pattern_name} ({matches/len(sample):.0%})")

        return patterns_found[:3]

    def _calculate_quality_score(self, columns: List[ColumnProfile]) -> float:
        """Calculate overall data quality score (0-100)."""
        if not columns:
            return 0.0

        scores = []

        for col in columns:
            col_score = 100

            # Penalize for nulls
            col_score -= min(col.null_percentage, 50)

            # Penalize for quality issues
            col_score -= len(col.quality_issues) * 10

            scores.append(max(col_score, 0))

        return round(sum(scores) / len(scores), 1)

    def _generate_recommendations(self, columns: List[ColumnProfile], df: pd.DataFrame) -> List[str]:
        """Generate recommendations based on profile."""
        recommendations = []

        # High null columns
        high_null = [c for c in columns if c.null_percentage > 30]
        if high_null:
            recommendations.append(
                f"Review {len(high_null)} columns with >30% null values: "
                f"{', '.join(c.name for c in high_null[:3])}"
            )

        # Potential ID columns without uniqueness
        for col in columns:
            if 'id' in col.name.lower() and col.uniqueness_ratio < 1.0:
                recommendations.append(
                    f"Column '{col.name}' appears to be an ID but has duplicate values"
                )

        # Date columns that should be datetime
        for col in columns:
            if col.inferred_type in ['date_iso', 'date_us'] and col.data_type == 'object':
                recommendations.append(
                    f"Convert '{col.name}' to datetime type for better analysis"
                )

        # Cost columns that are strings
        for col in columns:
            if col.inferred_type == 'currency' and col.data_type == 'object':
                recommendations.append(
                    f"Convert '{col.name}' to numeric type (remove $ and commas)"
                )

        return recommendations

    def profile_to_dict(self, profile: DataProfile) -> Dict:
        """Convert profile to dictionary for JSON export."""
        return {
            'source_name': profile.source_name,
            'row_count': profile.row_count,
            'column_count': profile.column_count,
            'duplicate_rows': profile.duplicate_rows,
            'memory_usage': profile.memory_usage,
            'profiled_at': profile.profiled_at.isoformat(),
            'quality_score': profile.quality_score,
            'recommendations': profile.recommendations,
            'columns': [
                {
                    'name': c.name,
                    'data_type': c.data_type,
                    'inferred_type': c.inferred_type,
                    'null_percentage': c.null_percentage,
                    'unique_count': c.unique_count,
                    'quality_issues': c.quality_issues,
                    'top_values': c.top_values[:3]
                }
                for c in profile.columns
            ]
        }

    def generate_profile_report(self, profile: DataProfile) -> str:
        """Generate markdown profile report."""
        report = [f"# Data Profile: {profile.source_name}", ""]
        report.append(f"**Profiled At:** {profile.profiled_at.strftime('%Y-%m-%d %H:%M')}")
        report.append(f"**Quality Score:** {profile.quality_score}/100")
        report.append("")

        # Summary
        report.append("## Summary")
        report.append(f"- **Rows:** {profile.row_count:,}")
        report.append(f"- **Columns:** {profile.column_count}")
        report.append(f"- **Duplicate Rows:** {profile.duplicate_rows:,}")
        report.append(f"- **Memory Usage:** {profile.memory_usage}")
        report.append("")

        # Recommendations
        if profile.recommendations:
            report.append("## Recommendations")
            for rec in profile.recommendations:
                report.append(f"- {rec}")
            report.append("")

        # Column Details
        report.append("## Column Details")
        report.append("")
        report.append("| Column | Type | Inferred | Nulls | Unique | Issues |")
        report.append("|--------|------|----------|-------|--------|--------|")

        for col in profile.columns:
            issues = len(col.quality_issues)
            report.append(
                f"| {col.name} | {col.data_type} | {col.inferred_type} | "
                f"{col.null_percentage:.1f}% | {col.unique_count:,} | {issues} |"
            )

        # Detailed column profiles
        report.append("")
        report.append("## Detailed Column Profiles")

        for col in profile.columns:
            report.append(f"\n### {col.name}")
            report.append(f"- **Type:** {col.data_type} (inferred: {col.inferred_type})")
            report.append(f"- **Nulls:** {col.null_count:,} ({col.null_percentage:.1f}%)")
            report.append(f"- **Unique Values:** {col.unique_count:,} ({col.uniqueness_ratio:.1%})")

            if col.min_value is not None:
                report.append(f"- **Range:** {col.min_value:,.2f} to {col.max_value:,.2f}")
                report.append(f"- **Mean:** {col.mean_value:,.2f}, Median: {col.median_value:,.2f}")

            if col.min_length is not None:
                report.append(f"- **Length:** {col.min_length} to {col.max_length} (avg: {col.avg_length:.1f})")

            if col.top_values:
                report.append(f"- **Top Values:** {col.top_values[:3]}")

            if col.common_patterns:
                report.append(f"- **Patterns:** {', '.join(col.common_patterns)}")

            if col.quality_issues:
                report.append(f"- **Issues:** {', '.join(col.quality_issues)}")

        return "\n".join(report)

    def compare_profiles(self, profile1: DataProfile, profile2: DataProfile) -> Dict:
        """Compare two profiles to detect schema changes or data drift."""
        comparison = {
            'profiles': [profile1.source_name, profile2.source_name],
            'row_count_change': profile2.row_count - profile1.row_count,
            'quality_change': profile2.quality_score - profile1.quality_score,
            'new_columns': [],
            'removed_columns': [],
            'type_changes': [],
            'null_rate_changes': []
        }

        cols1 = {c.name: c for c in profile1.columns}
        cols2 = {c.name: c for c in profile2.columns}

        # Find new/removed columns
        comparison['new_columns'] = [n for n in cols2 if n not in cols1]
        comparison['removed_columns'] = [n for n in cols1 if n not in cols2]

        # Compare common columns
        for name in cols1:
            if name in cols2:
                c1, c2 = cols1[name], cols2[name]

                if c1.data_type != c2.data_type:
                    comparison['type_changes'].append({
                        'column': name,
                        'from': c1.data_type,
                        'to': c2.data_type
                    })

                null_change = c2.null_percentage - c1.null_percentage
                if abs(null_change) > 10:
                    comparison['null_rate_changes'].append({
                        'column': name,
                        'change': null_change
                    })

        return comparison

クイックスタート

import pandas as pd

# Load construction data
df = pd.read_excel("project_costs.xlsx")

# Profile the data
profiler = ConstructionDataProfiler()
profile = profiler.profile_dataframe(df, "Project Costs 2025")

# Generate report
report = profiler.generate_profile_report(profile)
print(report)

# Export to JSON
profile_dict = profiler.profile_to_dict(profile)
with open("profile.json", "w") as f:
    json.dump(profile_dict, f, indent=2)

# Compare with previous profile
old_profile = profiler.profile_dataframe(old_df, "Project Costs 2024")
comparison = profiler.compare_profiles(old_profile, profile)
print(f"Quality changed by: {comparison['quality_change']}")

一般的なユースケース

1. ETL 前の分析：パイプラインを構築する前にソースデータをプロファイリングします
2. 品質の監視：データ品質を経時的に追跡します
3. スキーマの検証：データ構造に予期しない変更がないか検出します
4. 異常検出：外れ値やデータ品質の問題を見つけます

依存関係

pip install pandas numpy

リソース

• データプロファイリングのベストプラクティス：DAMA DMBOK
• 建設データの標準：CSI MasterFormat、UniFormat